github-actions[bot] commented on code in PR #65674:
URL: https://github.com/apache/doris/pull/65674#discussion_r3626889995
##########
be/src/format_v2/parquet/parquet_reader.cpp:
##########
@@ -779,13 +700,25 @@ Status ParquetReader::get_aggregate_result(const
format::FileAggregateRequest& r
auto& aggregate_column = result->columns[request_column_idx];
aggregate_column.projection =
request.columns[request_column_idx].projection;
for (const auto& row_group_plan : _state->scan_plan.row_groups) {
- auto row_group_metadata =
-
_state->file_context.metadata->RowGroup(row_group_plan.row_group_id);
- DORIS_CHECK(row_group_metadata != nullptr);
- auto column_chunk =
row_group_metadata->ColumnChunk(leaf_schema->leaf_column_id);
- DORIS_CHECK(column_chunk != nullptr);
+ const auto& row_group_metadata =
_state->file_context.native_metadata->to_thrift()
+
.row_groups[row_group_plan.row_group_id];
+ DORIS_CHECK(leaf_schema->leaf_column_id >= 0 &&
+ leaf_schema->leaf_column_id <
+
static_cast<int>(row_group_metadata.columns.size()));
+ const auto& column_chunk =
row_group_metadata.columns[leaf_schema->leaf_column_id];
+ DORIS_CHECK(column_chunk.__isset.meta_data);
+ const auto& column_metadata = column_chunk.meta_data;
+ std::optional<tparquet::Statistics> safe_statistics;
+ if (column_metadata.__isset.statistics) {
+ safe_statistics = detail::sanitize_native_footer_statistics(
Review Comment:
[P1] Do not return explicitly inexact footer bounds as MIN/MAX values
Parquet permits `min_value`/`max_value` to be compact bounds that are not
present in the chunk, and marks this with
`is_min_value_exact`/`is_max_value_exact`. This path sanitizes and returns
numeric bounds without consulting those flags (only binary physical types are
disabled), so a valid INT32 chunk whose real extrema are 1/99 but whose bounds
are 0/100 with both flags false returns 0/100. Treat an explicitly false
exactness flag as unavailable for aggregate pushdown (while preserving the
intended policy for absent legacy flags), and cover both numeric MIN and MAX.
##########
be/src/format_v2/parquet/parquet_file_context.cpp:
##########
@@ -15,154 +15,125 @@
#include "format_v2/parquet/parquet_file_context.h"
-#include <arrow/buffer.h>
-#include <arrow/result.h>
#include <fmt/format.h>
-#include <gen_cpp/segment_v2.pb.h>
-#include <parquet/exception.h>
#include <algorithm>
#include <cstring>
#include <exception>
#include <limits>
-#include <mutex>
#include <string_view>
#include <utility>
+#include "common/cast_set.h"
#include "common/check.h"
#include "common/config.h"
+#include "format_v2/parquet/parquet_statistics.h"
+#include "format_v2/parquet/reader/native/column_chunk_reader.h"
#include "io/cache/cached_remote_file_reader.h"
#include "io/file_factory.h"
#include "io/fs/buffered_reader.h"
+#include "io/fs/file_meta_cache.h"
#include "io/fs/file_reader.h"
#include "io/fs/tracing_file_reader.h"
#include "io/io_common.h"
-#include "storage/cache/page_cache.h"
+#include "runtime/exec_env.h"
+#include "util/coding.h"
#include "util/slice.h"
+#include "util/thrift_util.h"
+#include "util/time.h"
namespace doris::format::parquet {
-namespace detail {
-
-namespace {
-
-bool page_cache_range_less(const ParquetPageCacheRange& lhs, const
ParquetPageCacheRange& rhs) {
- return lhs.offset < rhs.offset || (lhs.offset == rhs.offset && lhs.size <
rhs.size);
-}
+constexpr size_t V2_PARQUET_FOOTER_SIZE = 8;
-} // namespace
-
-ParquetPageCacheRangeIndex::ParquetPageCacheRangeIndex(size_t max_ranges)
- : _max_ranges(max_ranges) {
- DORIS_CHECK(_max_ranges > 0);
+NativeParquetMetadata::NativeParquetMetadata(tparquet::FileMetaData metadata,
size_t parsed_size)
+ : _metadata(std::move(metadata)), _parsed_size(parsed_size) {
+ ExecEnv::GetInstance()->parquet_meta_tracker()->consume(get_mem_size());
}
-void ParquetPageCacheRangeIndex::insert(ParquetPageCacheRange range) {
- std::lock_guard lock(_mutex);
- auto it = std::lower_bound(_ranges.begin(), _ranges.end(), range,
page_cache_range_less);
- if (it != _ranges.end() && it->offset == range.offset && it->size ==
range.size) {
- return;
- }
- if (_ranges.size() >= _max_ranges) {
- _ranges.erase(_ranges.begin());
- it = std::lower_bound(_ranges.begin(), _ranges.end(), range,
page_cache_range_less);
- }
- _ranges.insert(it, range);
+NativeParquetMetadata::~NativeParquetMetadata() {
+ ExecEnv::GetInstance()->parquet_meta_tracker()->release(get_mem_size());
}
-void ParquetPageCacheRangeIndex::erase(ParquetPageCacheRange range) {
- std::lock_guard lock(_mutex);
- const auto it = std::lower_bound(_ranges.begin(), _ranges.end(), range,
page_cache_range_less);
- if (it != _ranges.end() && it->offset == range.offset && it->size ==
range.size) {
- _ranges.erase(it);
+Status NativeParquetMetadata::init_schema(bool enable_mapping_varbinary,
+ bool enable_mapping_timestamp_tz) {
+ _schema.set_enable_mapping_varbinary(enable_mapping_varbinary);
+ _schema.set_enable_mapping_timestamp_tz(enable_mapping_timestamp_tz);
+ RETURN_IF_ERROR(_schema.parse_from_thrift(_metadata.schema));
+ // Native readers address projected leaves by stable DFS IDs. Assign them
only on the private
+ // v2 schema object so v1's cached schema lifecycle and numbering remain
untouched.
+ _schema.assign_ids();
+ for (size_t row_group_idx = 0; row_group_idx <
_metadata.row_groups.size(); ++row_group_idx) {
+ const auto& row_group = _metadata.row_groups[row_group_idx];
+ if (row_group.num_rows < 0) {
+ return Status::Corruption("Parquet row group {} has negative row
count {}",
+ row_group_idx, row_group.num_rows);
+ }
+ if (row_group.columns.size() != _schema.physical_fields_size()) {
+ // All v2 planners index chunks by the native DFS leaf order, so
validate cardinality
+ // once before any projection, prefetch, or decoder can perform
indexed access.
+ return Status::Corruption(
+ "Parquet row group {} has {} column chunks but schema has
{} physical fields",
+ row_group_idx, row_group.columns.size(),
_schema.physical_fields_size());
+ }
+ for (size_t column_idx = 0; column_idx < row_group.columns.size();
++column_idx) {
+ const auto& chunk = row_group.columns[column_idx];
+ if (!chunk.__isset.meta_data) {
+ return Status::Corruption("Parquet row group {} column {} has
no metadata",
+ row_group_idx, column_idx);
+ }
+ if (chunk.meta_data.type !=
_schema.get_physical_field(column_idx)->physical_type) {
Review Comment:
[P1] Reconcile flat chunk value counts before footer pruning
For a repetition-level-zero leaf, `ColumnMetaData.num_values` includes NULL
slots and must equal `row_group.num_rows`. This metadata pass validates chunk
count and physical type but not that cardinality, while
`check_native_statistics()` later treats it as exact. A one-row OPTIONAL INT32
chunk with a real non-null row but `num_values=2,null_count=2` therefore
becomes an all-null zone map and lets `IS NOT NULL` prune the row group before
page validation can reject the footer. Reject the contradictory flat chunk (or
at minimum disable its statistics before pruning), and add this footer-pruning
case.
##########
be/src/format_v2/table_reader.h:
##########
@@ -602,7 +636,15 @@ class TableReader {
DORIS_CHECK(block->columns() > 0 || rows == 0);
for (size_t column_idx = 0; column_idx < block->columns();
++column_idx) {
auto column =
block->get_by_position(column_idx).type->create_column();
- column->resize(rows);
+ if (auto* nullable =
check_and_get_column<ColumnNullable>(*column)) {
+ // Metadata COUNT emits synthetic input rows for the unchanged
upper aggregate.
+ // They must be non-NULL for COUNT(nullable_col), and
constructing them explicitly
+ // also keeps every nullable null map boolean-valid in
debug/ASAN block checks.
Review Comment:
[P1] Bound synthetic COUNT materialization to the runtime batch
The newly enabled nullable `COUNT(col)` path passes the file's entire exact
`FileAggregateResult.count` here as `rows`. This branch then allocates that
many nested defaults plus a byte per row for the null map in one block,
although the unchanged upper COUNT immediately reduces those synthetic rows
again. A valid nullable BIGINT file with 500 million non-NULL rows can
therefore request roughly 4.5 GB before returning one scanner block. Keep a
remaining file-count state and emit at most one runtime batch per `get_block()`
call (as `_read_table_level_count()` already does), or disable this pushdown
until its result can be materialized boundedly; add a large synthetic-count
test that asserts the block cap.
##########
be/src/format_v2/parquet/native_schema_desc.cpp:
##########
@@ -0,0 +1,830 @@
+// 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 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 (...) {
+ // now the Not supported exception are ignored
+ // so those byte_array maybe be treated as varbinary(now) :
string(before)
+ }
+ if (ans.first->get_primitive_type() == PrimitiveType::INVALID_TYPE) {
+ switch (physical_schema.type) {
+ case tparquet::Type::BOOLEAN:
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_BOOLEAN, nullable);
+ break;
+ case tparquet::Type::INT32:
+ ans.first = DataTypeFactory::instance().create_data_type(TYPE_INT,
nullable);
+ break;
+ case tparquet::Type::INT64:
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_BIGINT, nullable);
+ break;
+ case tparquet::Type::INT96:
+ if (_enable_mapping_timestamp_tz) {
+ // treat INT96 as TIMESTAMPTZ
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_TIMESTAMPTZ, nullable,
+ 0, 6);
+ } else {
+ // in most cases, it's a nano timestamp
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_DATETIMEV2, nullable,
+ 0, 6);
+ }
+ break;
+ case tparquet::Type::FLOAT:
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_FLOAT, nullable);
+ break;
+ case tparquet::Type::DOUBLE:
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_DOUBLE, nullable);
+ break;
+ case tparquet::Type::BYTE_ARRAY:
+ if (_enable_mapping_varbinary) {
+ // if physical_schema not set logicalType and converted_type,
+ // we treat BYTE_ARRAY as VARBINARY by default, so that we can
read all data directly.
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_VARBINARY, nullable);
+ } else {
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_STRING, nullable);
+ }
+ break;
+ case tparquet::Type::FIXED_LEN_BYTE_ARRAY:
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_STRING, nullable);
+ break;
+ default:
+ throw Exception(Status::InternalError("Not supported parquet
logicalType{}",
+ physical_schema.type));
+ break;
+ }
+ }
+ return ans;
+}
+
+std::pair<DataTypePtr, bool> NativeFieldDescriptor::convert_to_doris_type(
+ tparquet::LogicalType logicalType, bool nullable) {
+ std::pair<DataTypePtr, bool> ans = {std::make_shared<DataTypeNothing>(),
false};
+ bool& is_type_compatibility = ans.second;
+ if (logicalType.__isset.STRING || logicalType.__isset.ENUM ||
logicalType.__isset.JSON ||
+ logicalType.__isset.BSON) {
+ ans.first = DataTypeFactory::instance().create_data_type(TYPE_STRING,
nullable);
+ } else if (logicalType.__isset.DECIMAL) {
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_DECIMAL128I, nullable,
+
logicalType.DECIMAL.precision,
+
logicalType.DECIMAL.scale);
+ } else if (logicalType.__isset.DATE) {
+ ans.first = DataTypeFactory::instance().create_data_type(TYPE_DATEV2,
nullable);
+ } else if (logicalType.__isset.INTEGER) {
+ if (logicalType.INTEGER.isSigned) {
+ if (logicalType.INTEGER.bitWidth <= 8) {
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_TINYINT, nullable);
+ } else if (logicalType.INTEGER.bitWidth <= 16) {
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_SMALLINT, nullable);
+ } else if (logicalType.INTEGER.bitWidth <= 32) {
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_INT, nullable);
+ } else {
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_BIGINT, nullable);
+ }
+ } else {
+ is_type_compatibility = true;
+ if (logicalType.INTEGER.bitWidth <= 8) {
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_SMALLINT, nullable);
+ } else if (logicalType.INTEGER.bitWidth <= 16) {
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_INT, nullable);
+ } else if (logicalType.INTEGER.bitWidth <= 32) {
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_BIGINT, nullable);
+ } else {
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_LARGEINT, nullable);
+ }
+ }
+ } else if (logicalType.__isset.TIME) {
+ const int scale = logicalType.TIME.unit.__isset.MILLIS ? 3 : 6;
+ // TIME stores an integer unit, so its Doris scale must preserve the
footer unit or
+ // sub-second values are silently truncated by the target SerDe.
+ ans.first = DataTypeFactory::instance().create_data_type(TYPE_TIMEV2,
nullable, 0, scale);
+ } else if (logicalType.__isset.TIMESTAMP) {
+ if (_enable_mapping_timestamp_tz) {
+ if (logicalType.TIMESTAMP.isAdjustedToUTC) {
+ // treat TIMESTAMP with isAdjustedToUTC as TIMESTAMPTZ
+ ans.first = DataTypeFactory::instance().create_data_type(
+ TYPE_TIMESTAMPTZ, nullable, 0,
+ logicalType.TIMESTAMP.unit.__isset.MILLIS ? 3 : 6);
+ return ans;
+ }
+ }
+ ans.first = DataTypeFactory::instance().create_data_type(
+ TYPE_DATETIMEV2, nullable, 0,
logicalType.TIMESTAMP.unit.__isset.MILLIS ? 3 : 6);
+ } else if (logicalType.__isset.UUID) {
+ if (_enable_mapping_varbinary) {
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_VARBINARY, nullable, -1,
+ -1, 16);
+ } else {
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_STRING, nullable);
+ }
+ } else if (logicalType.__isset.FLOAT16) {
+ ans.first = DataTypeFactory::instance().create_data_type(TYPE_FLOAT,
nullable);
+ } else {
+ throw Exception(Status::InternalError("Not supported parquet
logicalType"));
+ }
+ return ans;
+}
+
+std::pair<DataTypePtr, bool> NativeFieldDescriptor::convert_to_doris_type(
+ const tparquet::SchemaElement& physical_schema, bool nullable) {
+ std::pair<DataTypePtr, bool> ans = {std::make_shared<DataTypeNothing>(),
false};
+ bool& is_type_compatibility = ans.second;
+ switch (physical_schema.converted_type) {
+ case tparquet::ConvertedType::type::UTF8:
+ case tparquet::ConvertedType::type::ENUM:
+ case tparquet::ConvertedType::type::JSON:
+ case tparquet::ConvertedType::type::BSON:
+ ans.first = DataTypeFactory::instance().create_data_type(TYPE_STRING,
nullable);
+ break;
+ case tparquet::ConvertedType::type::DECIMAL:
+ ans.first = DataTypeFactory::instance().create_data_type(
+ TYPE_DECIMAL128I, nullable, physical_schema.precision,
physical_schema.scale);
+ break;
+ case tparquet::ConvertedType::type::DATE:
+ ans.first = DataTypeFactory::instance().create_data_type(TYPE_DATEV2,
nullable);
+ break;
+ case tparquet::ConvertedType::type::TIME_MILLIS:
+ ans.first = DataTypeFactory::instance().create_data_type(TYPE_TIMEV2,
nullable, 0, 3);
+ break;
+ case tparquet::ConvertedType::type::TIME_MICROS:
+ ans.first = DataTypeFactory::instance().create_data_type(TYPE_TIMEV2,
nullable, 0, 6);
+ break;
+ case tparquet::ConvertedType::type::TIMESTAMP_MILLIS:
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_DATETIMEV2, nullable, 0, 3);
+ break;
+ case tparquet::ConvertedType::type::TIMESTAMP_MICROS:
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_DATETIMEV2, nullable, 0, 6);
+ break;
+ case tparquet::ConvertedType::type::INT_8:
+ ans.first = DataTypeFactory::instance().create_data_type(TYPE_TINYINT,
nullable);
+ break;
+ case tparquet::ConvertedType::type::UINT_8:
+ is_type_compatibility = true;
+ [[fallthrough]];
+ case tparquet::ConvertedType::type::INT_16:
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_SMALLINT, nullable);
+ break;
+ case tparquet::ConvertedType::type::UINT_16:
+ is_type_compatibility = true;
+ [[fallthrough]];
+ case tparquet::ConvertedType::type::INT_32:
+ ans.first = DataTypeFactory::instance().create_data_type(TYPE_INT,
nullable);
+ break;
+ case tparquet::ConvertedType::type::UINT_32:
+ is_type_compatibility = true;
+ [[fallthrough]];
+ case tparquet::ConvertedType::type::INT_64:
+ ans.first = DataTypeFactory::instance().create_data_type(TYPE_BIGINT,
nullable);
+ break;
+ case tparquet::ConvertedType::type::UINT_64:
+ is_type_compatibility = true;
+ ans.first =
DataTypeFactory::instance().create_data_type(TYPE_LARGEINT, nullable);
+ break;
+ default:
+ throw Exception(Status::InternalError("Not supported parquet
ConvertedType: {}",
+ physical_schema.converted_type));
+ }
+ return ans;
+}
+
+Status NativeFieldDescriptor::parse_group_field(
+ const std::vector<tparquet::SchemaElement>& t_schemas, size_t curr_pos,
+ NativeFieldSchema* group_field) {
+ auto& group_schema = t_schemas[curr_pos];
+ if ((group_schema.__isset.logicalType &&
group_schema.logicalType.__isset.ENUM) ||
Review Comment:
[P2] Reject all primitive-only annotations on group nodes
This special case rejects only ENUM. A group annotated STRING, DATE,
DECIMAL, TIME/TIMESTAMP, INTEGER, UUID, FLOAT16, or a corresponding legacy
primitive annotation still falls through to `parse_struct_field()` and is
exposed as an ordinary STRUCT; the leaf physical-annotation validator cannot
catch it because this node has no physical carrier. Reject every known
primitive-only annotation before generic group parsing (while retaining the
supported LIST/MAP forms), and parameterize the malformed-group test beyond
ENUM.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/implementation/AggregateStrategies.java:
##########
@@ -740,10 +740,15 @@ private LogicalAggregate<? extends Plan>
storageLayerAggregate(
}
}
if (mergeOp == PushDownAggOp.COUNT || mergeOp ==
PushDownAggOp.MIX) {
- // NULL value behavior in `count` function is zero, so
- // we should not use row_count to speed up query. the col
- // must be not null
- if (column.isAllowNull() && checkNullSlots.contains(slot)) {
+ // Nullable file COUNT is exact only when this query is routed
to FileScannerV2,
+ // which carries the semantic argument and counts definition
levels. Gating on the
+ // session switch keeps V1 on its original full-column
evaluation path.
+ boolean supportsNullableFileCount = logicalScan instanceof
LogicalFileScan
+ && mergeOp == PushDownAggOp.COUNT
Review Comment:
[P1] Preserve distinct COUNT cardinalities in mixed aggregates
A query containing both `COUNT(*)` and `COUNT(nullable_col)` reaches this
gate with `mergeOp == COUNT`, but only the nullable argument is serialized in
`countArgumentExprIds`; the zero-argument COUNT contributes no slot.
Parquet/ORC then returns only `COUNT(nullable_col)`, TableReader emits that
many non-NULL synthetic rows, and the unchanged upper `COUNT(*)` and
`COUNT(col)` both count them. Thus `[NULL, 1]` returns `(1, 1)` instead of `(2,
1)`. Allow file COUNT pushdown only when every upper COUNT has the same
effective cardinality, or carry separate results, and add this mixed aggregate
case.
--
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]
