github-actions[bot] commented on code in PR #65674:
URL: https://github.com/apache/doris/pull/65674#discussion_r3608402630
##########
be/src/format_v2/parquet/parquet_scan.cpp:
##########
@@ -201,19 +261,25 @@ std::vector<ParquetPageCacheRange>
build_row_group_prefetch_ranges(
collect_projected_leaf_column_ids(*file_schema[local_id], projection,
&leaf_column_ids);
}
- auto row_group_metadata = metadata.RowGroup(row_group_idx);
- DORIS_CHECK(row_group_metadata != nullptr);
+ DORIS_CHECK(row_group_idx >= 0 && row_group_idx <
static_cast<int>(metadata.row_groups.size()));
+ const auto& row_group_metadata = metadata.row_groups[row_group_idx];
std::vector<int> ordered_leaf_column_ids(leaf_column_ids.begin(),
leaf_column_ids.end());
std::ranges::sort(ordered_leaf_column_ids);
std::vector<ParquetPageCacheRange> ranges;
ranges.reserve(ordered_leaf_column_ids.size());
for (const auto leaf_column_id : ordered_leaf_column_ids) {
- DORIS_CHECK(leaf_column_id >= 0 && leaf_column_id <
row_group_metadata->num_columns());
- auto column_metadata = row_group_metadata->ColumnChunk(leaf_column_id);
- DORIS_CHECK(column_metadata != nullptr);
- const int64_t offset = column_start_offset(*column_metadata);
- const int64_t size = column_metadata->total_compressed_size();
+ DORIS_CHECK(leaf_column_id >= 0 &&
Review Comment:
[P1] Reject row groups that cannot satisfy the schema
This assertion consumes footer-controlled structure before any ordinary
column reader validates it. For example, a two-leaf schema with a positive-row
row group containing only one `ColumnChunk` reaches this path on a no-predicate
scan of the second leaf and aborts the BE; `native::ColumnReader::create()`
also retains an unchecked `row_group.columns[physical_index]` access. Please
validate every row group's chunk cardinality and required metadata against the
physical schema during native metadata initialization, keep recoverable bounds
checks at consumers, and add malformed row-group tests with and without pruning
predicates.
##########
be/src/format_v2/parquet/parquet_scan.cpp:
##########
@@ -201,19 +261,25 @@ std::vector<ParquetPageCacheRange>
build_row_group_prefetch_ranges(
collect_projected_leaf_column_ids(*file_schema[local_id], projection,
&leaf_column_ids);
}
- auto row_group_metadata = metadata.RowGroup(row_group_idx);
- DORIS_CHECK(row_group_metadata != nullptr);
+ DORIS_CHECK(row_group_idx >= 0 && row_group_idx <
static_cast<int>(metadata.row_groups.size()));
+ const auto& row_group_metadata = metadata.row_groups[row_group_idx];
std::vector<int> ordered_leaf_column_ids(leaf_column_ids.begin(),
leaf_column_ids.end());
std::ranges::sort(ordered_leaf_column_ids);
std::vector<ParquetPageCacheRange> ranges;
ranges.reserve(ordered_leaf_column_ids.size());
for (const auto leaf_column_id : ordered_leaf_column_ids) {
- DORIS_CHECK(leaf_column_id >= 0 && leaf_column_id <
row_group_metadata->num_columns());
- auto column_metadata = row_group_metadata->ColumnChunk(leaf_column_id);
- DORIS_CHECK(column_metadata != nullptr);
- const int64_t offset = column_start_offset(*column_metadata);
- const int64_t size = column_metadata->total_compressed_size();
+ DORIS_CHECK(leaf_column_id >= 0 &&
+ leaf_column_id <
static_cast<int>(row_group_metadata.columns.size()));
+ const auto& chunk = row_group_metadata.columns[leaf_column_id];
+ if (!chunk.__isset.meta_data) {
+ continue;
+ }
+ const auto& column_metadata = chunk.meta_data;
+ const int64_t offset = column_metadata.__isset.dictionary_page_offset
Review Comment:
[P1] Build prefetch ranges from the checked chunk range
This bypasses the signed/file-bounded `compute_column_chunk_range()`
contract fixed in the earlier thread. If the footer sets
`dictionary_page_offset = -1` alongside a valid data-page offset, the checked
reader helper deliberately falls back to the data offset, but this raw `isset`
selection reaches `DORIS_CHECK(offset >= 0)` and aborts first because row-group
setup invokes it before reader construction. Make range construction return
`Status` and reuse the validated chunk range (also covering
overflow/out-of-file extents) instead of asserting on raw optional metadata;
add normal-open and prefetch tests for invalid dictionary/data offsets.
##########
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:
[P1] Validate logical annotations against their physical type
The native path records the annotation independently of the physical storage
and only validates UUID/FLOAT16. A BOOLEAN leaf annotated DATE, for example, is
exposed as DATEV2; the Bool decoder then supplies one-byte values to
`DateV2ParquetConsumer`, whose width-four `DORIS_CHECK` aborts the BE.
TIME/TIMESTAMP have analogous fatal pairings, while invalid INTEGER pairings
can be silently interpreted. Validate the complete Parquet
logical/converted-type storage matrix (including fixed-length widths) during
schema/context initialization and return corruption before publishing a
decoder, with dense/dictionary/selected tests for disallowed pairs.
##########
be/src/format_v2/parquet/parquet_statistics.cpp:
##########
@@ -47,23 +47,70 @@
#include "core/field.h"
#include "exprs/expr_zonemap_filter.h"
#include "exprs/vexpr_context.h"
+#include "format/parquet/parquet_block_split_bloom_filter.h"
#include "format_v2/parquet/parquet_column_schema.h"
+#include "format_v2/parquet/parquet_file_context.h"
+#include "format_v2/parquet/reader/native_column_reader.h"
#include "format_v2/timestamp_statistics.h"
#include "runtime/runtime_profile.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 {
+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);
+
enum class ParquetRowGroupPruneReason {
NONE, // cannot prune; must read
STATISTICS, // excluded by ZoneMap statistics
DICTIONARY, // excluded by dictionary
BLOOM_FILTER, // excluded by bloom filter
};
+Status read_native_bloom_filter(const tparquet::ColumnMetaData& metadata,
+ const io::FileReaderSPtr& file, io::IOContext*
io_ctx,
+ std::unique_ptr<ParquetBlockSplitBloomFilter>*
result) {
+ if (result == nullptr || file == nullptr ||
!metadata.__isset.bloom_filter_offset) {
+ return Status::NotSupported("Parquet Bloom filter is unavailable");
+ }
+ constexpr size_t MAX_BLOOM_HEADER_BYTES = 64;
+ const size_t header_read_size =
+ metadata.__isset.bloom_filter_length &&
metadata.bloom_filter_length > 0
+ ? std::min<size_t>(metadata.bloom_filter_length,
MAX_BLOOM_HEADER_BYTES)
+ : MAX_BLOOM_HEADER_BYTES;
+ std::vector<uint8_t> header_buffer(header_read_size);
+ size_t bytes_read = 0;
+ RETURN_IF_ERROR(file->read_at(metadata.bloom_filter_offset,
+ Slice(header_buffer.data(),
header_buffer.size()), &bytes_read,
+ io_ctx));
+ tparquet::BloomFilterHeader header;
+ uint32_t header_size = cast_set<uint32_t>(bytes_read);
+ RETURN_IF_ERROR(deserialize_thrift_msg(header_buffer.data(), &header_size,
true, &header));
+ if (!header.algorithm.__isset.BLOCK ||
!header.compression.__isset.UNCOMPRESSED ||
+ !header.hash.__isset.XXHASH || header.numBytes <= 0) {
+ return Status::NotSupported("Unsupported Parquet Bloom filter
encoding");
+ }
+
+ std::vector<uint8_t> data(cast_set<size_t>(header.numBytes));
Review Comment:
[P1] Validate the Bloom payload before allocating it
`header.numBytes` is file-controlled, but every positive value is accepted
here. A tiny filter with `numBytes = 2` passes `init()` and is then probed by
`test_hash()`, which unconditionally reads one 32-byte split block past that
allocation. At the other extreme, `INT32_MAX` requests roughly 2 GiB before
`read_at()` can reject the absent payload, and a successful load deep-copies
it. `bloom_filter_length` currently only caps the header read; it never bounds
`header_size + numBytes` or the remaining file extent. Please require complete
32-byte blocks, apply the existing 128 MiB Bloom maximum, validate the
declared/file range with checked arithmetic, and fall back from this optional
pruning metadata, with sub-block and huge-header tests.
##########
be/src/format_v2/parquet/parquet_statistics.cpp:
##########
@@ -815,6 +867,43 @@ ParquetColumnStatistics
ParquetStatisticsUtils::TransformColumnStatistics(
}
}
+ParquetColumnStatistics ParquetStatisticsUtils::TransformColumnStatistics(
+ const ParquetColumnSchema& column_schema, const tparquet::Statistics*
statistics,
+ int64_t column_value_count, const cctz::time_zone* timezone) {
+ ParquetColumnStatistics result;
+ if (statistics == nullptr || column_value_count < 0) {
+ return result;
+ }
+
+ const bool has_null_count = statistics->__isset.null_count &&
statistics->null_count >= 0;
Review Comment:
[P1] Reject footer null counts above the value count
Any nonnegative `null_count` is treated as usable here. If a malformed
footer advertises `num_values = 1` and `null_count = 2`, this synthesizes
`null_pages = true`, and row-group pruning can discard an actual non-null row
for `IS NOT NULL` before reading data. This is separate from the ColumnIndex
consistency thread because the contradiction is introduced while converting
footer statistics and `column_value_count` is already available. Conservatively
disable these optional statistics when `null_count > column_value_count`, with
`IS NULL`/`IS NOT NULL` tests.
##########
be/src/format_v2/parquet/parquet_scan.cpp:
##########
@@ -331,7 +428,137 @@ Status plan_parquet_row_groups(const
::parquet::FileMetaData& metadata,
RETURN_IF_ERROR(build_row_group_read_plans(metadata, file_reader,
file_schema, request,
metadata_selected_row_groups,
row_group_first_rows,
- plan, timezone, runtime_state));
+ plan, timezone, runtime_state,
file_context));
+ plan->pruning_stats.selected_row_groups = plan->row_groups.size();
+ return Status::OK();
+}
+
+namespace {
+
+int64_t native_column_start_offset(const tparquet::ColumnMetaData&
column_metadata) {
Review Comment:
[P1] Validate chunk offsets before assigning scan splits
Split selection consumes this raw optional offset before any checked reader
range exists. With a valid data chunk at offset 500/size 100 but
`dictionary_page_offset = -1000`, a one-column row group gets midpoint -950, so
every non-full split excludes it and the query can succeed with those rows
silently missing. `compute_column_chunk_range()` would ignore that nonpositive
dictionary offset, but it is never reached. Derive split extents through the
same status-returning, file-bounded compatibility contract, use checked
midpoint arithmetic, and test that invalid/legacy offsets assign each row group
exactly once or return corruption.
##########
be/src/format_v2/parquet/native_schema_desc.cpp:
##########
@@ -0,0 +1,729 @@
+// 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) {
+ return schema.__isset.converted_type && schema.converted_type ==
tparquet::ConvertedType::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);
+}
+
+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;
+}
+
+/**
+ * `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& name = schema.name;
+ static const Slice array_slice("array", 5);
+ static const Slice tuple_slice("_tuple", 6);
+ Slice slice(name);
+ return slice == array_slice || slice.ends_with(tuple_slice);
+}
+
+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) {
+ if (t_schemas.size() == 0 || !is_group_node(t_schemas[0])) {
+ return Status::InvalidArgument("Wrong parquet root schema element");
+ }
+ 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) {
+ 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) {
+ ans.first = DataTypeFactory::instance().create_data_type(TYPE_TIMEV2,
nullable);
+ } 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.JSON) {
+ ans.first = DataTypeFactory::instance().create_data_type(TYPE_STRING,
nullable);
+ } 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:
+ 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:
+ [[fallthrough]];
+ case tparquet::ConvertedType::type::TIME_MICROS:
+ ans.first = DataTypeFactory::instance().create_data_type(TYPE_TIMEV2,
nullable);
+ 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;
+ case tparquet::ConvertedType::type::JSON:
+ ans.first = DataTypeFactory::instance().create_data_type(TYPE_STRING,
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) ||
+ (group_schema.__isset.converted_type &&
+ group_schema.converted_type == tparquet::ConvertedType::ENUM)) {
+ // ENUM describes primitive bytes only. Rejecting it before recursive
group parsing keeps
+ // the native metadata tree from silently accepting a schema the
Parquet contract forbids.
+ return Status::InvalidArgument("Logical type Enum cannot be applied to
group node");
+ }
+ if (is_map_node(group_schema)) {
+ // the map definition:
+ // optional group <name> (MAP) {
+ // repeated group map (MAP_KEY_VALUE) {
+ // required <type> key;
+ // optional <type> value;
+ // }
+ // }
+ return parse_map_field(t_schemas, curr_pos, group_field);
+ }
+ if (is_list_node(group_schema)) {
+ // the list definition:
+ // optional group <name> (LIST) {
+ // repeated group [bag | list] { // hive or spark
+ // optional <type> [array_element | element]; // hive or spark
+ // }
+ // }
+ return parse_list_field(t_schemas, curr_pos, group_field);
+ }
+
+ if (is_repeated_node(group_schema)) {
+ group_field->repetition_level++;
+ group_field->definition_level++;
+ group_field->children.resize(1);
+ set_child_node_level(group_field, group_field->definition_level);
+ auto struct_field = &group_field->children[0];
+ // the list of struct:
+ // repeated group <name> (LIST) {
+ // optional/required <type> <name>;
+ // ...
+ // }
+ // produce a non-null list<struct>
+ RETURN_IF_ERROR(parse_struct_field(t_schemas, curr_pos, struct_field));
+
+ group_field->name = group_schema.name;
+ group_field->lower_case_name = to_lower(group_field->name);
+ group_field->column_id = NATIVE_UNASSIGNED_COLUMN_ID; // Initialize
column_id
+ group_field->data_type =
+
std::make_shared<DataTypeArray>(make_nullable(struct_field->data_type));
+ group_field->field_id = group_schema.__isset.field_id ?
group_schema.field_id : -1;
+ } else {
+ RETURN_IF_ERROR(parse_struct_field(t_schemas, curr_pos, group_field));
+ }
+
+ return Status::OK();
+}
+
+Status NativeFieldDescriptor::parse_list_field(
+ const std::vector<tparquet::SchemaElement>& t_schemas, size_t curr_pos,
+ NativeFieldSchema* list_field) {
+ // the list definition:
+ // spark and hive have three level schemas but with different schema name
+ // spark: <column-name> - "list" - "element"
+ // hive: <column-name> - "bag" - "array_element"
+ // parse three level schemas to two level primitive like: LIST<INT>,
+ // or nested structure like: LIST<MAP<INT, INT>>
+ auto& first_level = t_schemas[curr_pos];
+ if (first_level.num_children != 1) {
+ return Status::InvalidArgument("List element should have only one
child");
+ }
+
+ if (curr_pos + 1 >= t_schemas.size()) {
+ return Status::InvalidArgument("List element should have the second
level schema");
+ }
+
+ if (first_level.repetition_type ==
tparquet::FieldRepetitionType::REPEATED) {
+ return Status::InvalidArgument("List element can't be a repeated
schema");
+ }
+
+ // the repeated schema element
+ auto& second_level = t_schemas[curr_pos + 1];
+ if (second_level.repetition_type !=
tparquet::FieldRepetitionType::REPEATED) {
+ return Status::InvalidArgument("The second level of list element
should be repeated");
+ }
+
+ // This indicates if this list is nullable.
+ bool is_optional = is_optional_node(first_level);
+ if (is_optional) {
+ list_field->definition_level++;
+ }
+ list_field->repetition_level++;
+ list_field->definition_level++;
+ list_field->children.resize(1);
+ NativeFieldSchema* list_child = &list_field->children[0];
+
+ size_t num_children = num_children_node(second_level);
+ if (num_children > 0) {
+ if (num_children == 1 && !is_struct_list_node(second_level)) {
+ // optional field, and the third level element is the nested
structure in list
+ // produce nested structure like: LIST<INT>, LIST<MAP>,
LIST<LIST<...>>
+ // skip bag/list, it's a repeated element.
+ set_child_node_level(list_field, list_field->definition_level);
+ RETURN_IF_ERROR(parse_node_field(t_schemas, curr_pos + 2,
list_child));
+ } else {
+ // required field, produce the list of struct
+ set_child_node_level(list_field, list_field->definition_level);
+ RETURN_IF_ERROR(parse_struct_field(t_schemas, curr_pos + 1,
list_child));
+ }
+ } else if (num_children == 0) {
+ // required two level list, for compatibility reason.
+ set_child_node_level(list_field, list_field->definition_level);
+ parse_physical_field(second_level, false, list_child);
+ _next_schema_pos = curr_pos + 2;
+ }
+
+ list_field->name = first_level.name;
+ list_field->lower_case_name = to_lower(first_level.name);
+ list_field->column_id = NATIVE_UNASSIGNED_COLUMN_ID; // Initialize
column_id
+ list_field->data_type =
+
std::make_shared<DataTypeArray>(make_nullable(list_field->children[0].data_type));
+ if (is_optional) {
+ list_field->data_type = make_nullable(list_field->data_type);
+ }
+ list_field->field_id = first_level.__isset.field_id ? first_level.field_id
: -1;
+
+ return Status::OK();
+}
+
+Status NativeFieldDescriptor::parse_map_field(const
std::vector<tparquet::SchemaElement>& t_schemas,
+ size_t curr_pos,
NativeFieldSchema* map_field) {
+ // the map definition in parquet:
+ // optional group <name> (MAP) {
+ // repeated group map (MAP_KEY_VALUE) {
+ // required <type> key;
+ // optional <type> value;
+ // }
+ // }
+ // Map value can be optional, the map without values is a SET
+ if (curr_pos + 2 >= t_schemas.size()) {
+ return Status::InvalidArgument("Map element should have at least three
levels");
+ }
+ auto& map_schema = t_schemas[curr_pos];
+ if (map_schema.num_children != 1) {
+ return Status::InvalidArgument(
+ "Map element should have only one child(name='map',
type='MAP_KEY_VALUE')");
+ }
+ if (is_repeated_node(map_schema)) {
+ return Status::InvalidArgument("Map element can't be a repeated
schema");
+ }
+ auto& map_key_value = t_schemas[curr_pos + 1];
+ if (!is_group_node(map_key_value) || !is_repeated_node(map_key_value)) {
+ return Status::InvalidArgument(
+ "the second level in map must be a repeated group(key and
value)");
+ }
+ auto& map_key = t_schemas[curr_pos + 2];
+ if (!is_required_node(map_key)) {
+ LOG(WARNING) << "Filed " << map_schema.name << " is map type, but with
nullable key column";
+ }
+
+ if (map_key_value.num_children == 1) {
+ // The map with three levels is a SET
+ return parse_list_field(t_schemas, curr_pos, map_field);
+ }
+ if (map_key_value.num_children != 2) {
+ // A standard map should have four levels
+ return Status::InvalidArgument(
+ "the second level in map(MAP_KEY_VALUE) should have two
children");
+ }
+ // standard map
+ bool is_optional = is_optional_node(map_schema);
+ if (is_optional) {
+ map_field->definition_level++;
+ }
+ map_field->repetition_level++;
+ map_field->definition_level++;
+
+ // Directly create key and value children instead of intermediate
key_value node
+ map_field->children.resize(2);
+ // map is a repeated node, we should set the `repeated_parent_def_level`
of its children as `definition_level`
+ set_child_node_level(map_field, map_field->definition_level);
+
+ auto key_field = &map_field->children[0];
+ auto value_field = &map_field->children[1];
+
+ // Parse key and value fields directly from the key_value group's children
+ _next_schema_pos = curr_pos + 2; // Skip key_value group, go directly to
key
+ RETURN_IF_ERROR(parse_node_field(t_schemas, _next_schema_pos, key_field));
+ RETURN_IF_ERROR(parse_node_field(t_schemas, _next_schema_pos,
value_field));
+
+ map_field->name = map_schema.name;
+ map_field->lower_case_name = to_lower(map_field->name);
+ map_field->column_id = NATIVE_UNASSIGNED_COLUMN_ID; // Initialize column_id
+ map_field->data_type =
std::make_shared<DataTypeMap>(make_nullable(key_field->data_type),
+
make_nullable(value_field->data_type));
+ if (is_optional) {
+ map_field->data_type = make_nullable(map_field->data_type);
+ }
+ map_field->field_id = map_schema.__isset.field_id ? map_schema.field_id :
-1;
+
+ return Status::OK();
+}
+
+Status NativeFieldDescriptor::parse_struct_field(
+ const std::vector<tparquet::SchemaElement>& t_schemas, size_t curr_pos,
+ NativeFieldSchema* struct_field) {
+ // the nested column in parquet, parse group to struct.
+ auto& struct_schema = t_schemas[curr_pos];
+ bool is_optional = is_optional_node(struct_schema);
+ if (is_optional) {
+ struct_field->definition_level++;
+ }
+ auto num_children = struct_schema.num_children;
+ struct_field->children.resize(num_children);
+ set_child_node_level(struct_field,
struct_field->repeated_parent_def_level);
+ _next_schema_pos = curr_pos + 1;
+ for (int i = 0; i < num_children; ++i) {
+ RETURN_IF_ERROR(parse_node_field(t_schemas, _next_schema_pos,
&struct_field->children[i]));
Review Comment:
[P1] Bound schema nesting before recursive parsing
Even after bounding each `num_children`, a flat footer can encode an
arbitrarily deep chain of REQUIRED groups with exactly one child. This call
consumes one C++ stack frame per `SchemaElement`, and `assign_ids()`
immediately recurses through the same tree again, so a modest-size deeply
nested footer can stack-overflow the BE while every local child count is valid.
Pass and enforce a conservative maximum depth during parsing (and keep later
recursive traversals within that admitted limit), with just-below/above-limit
one-child STRUCT tests.
##########
regression-test/data/nereids_rules_p0/pkfk/eliminate_inner.out:
##########
@@ -144,7 +144,7 @@ PhysicalResultSink
--NestedLoopJoin[INNER_JOIN]
----filter((pkt.pk = 1))
------PhysicalOlapScan[pkt]
-----filter((cast(f as DECIMALV3(38, 6)) = 1.000000) and (fkt_not_null.fk = 1))
+----filter((cast(fkt_not_null.f as DECIMALV3(38, 6)) = 1.000000) and
(fkt_not_null.fk = 1))
Review Comment:
[P1] Revert the unrelated Nereids golden-output edits
This PR changes only the expected qualification of FE plan slots in
`eliminate_inner.out`; it contains no FE implementation change and no change to
the corresponding Groovy input. The BE Parquet scanner cannot affect Nereids
`EXPLAIN SHAPE PLAN` rendering, so regenerating this suite from the PR code
will still emit the base output and fail against these expectations. Please
revert this unrelated golden-file hunk, or include the actual FE producer
change and regenerate the suite from it.
##########
be/src/format_v2/parquet/native_schema_desc.cpp:
##########
@@ -0,0 +1,729 @@
+// 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) {
+ return schema.__isset.converted_type && schema.converted_type ==
tparquet::ConvertedType::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);
+}
+
+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;
+}
+
+/**
+ * `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& name = schema.name;
+ static const Slice array_slice("array", 5);
+ static const Slice tuple_slice("_tuple", 6);
+ Slice slice(name);
+ return slice == array_slice || slice.ends_with(tuple_slice);
+}
+
+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) {
+ if (t_schemas.size() == 0 || !is_group_node(t_schemas[0])) {
+ return Status::InvalidArgument("Wrong parquet root schema element");
+ }
+ const auto& root_schema = t_schemas[0];
+ _fields.resize(root_schema.num_children);
Review Comment:
[P1] Bound schema child counts before resizing
`num_children` comes from the footer, but this resize happens before
checking that enough `SchemaElement`s remain. A tiny footer containing only a
root with `num_children = INT32_MAX` therefore attempts an enormous
`NativeFieldSchema` allocation instead of returning the later out-of-bounds
status; `parse_struct_field()` repeats the same allocate-before-validation
pattern. Require each positive child count to fit the remaining schema-element
budget before either resize, and add root/nested STRUCT huge-count tests.
##########
be/src/format_v2/parquet/reader/native/column_chunk_reader.cpp:
##########
@@ -0,0 +1,1289 @@
+// 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_chunk_reader.h"
+
+#include <cctz/time_zone.h>
+#include <gen_cpp/parquet_types.h>
+#include <glog/logging.h>
+#include <parquet/metadata.h>
+#include <string.h>
+
+#include <algorithm>
+#include <cstdint>
+#include <limits>
+#include <memory>
+#include <utility>
+
+#include "common/compiler_util.h" // IWYU pragma: keep
+#include "core/column/column.h"
+#include "core/column/column_vector.h"
+#include "core/custom_allocator.h"
+#include "core/data_type_serde/data_type_serde.h"
+#include "core/data_type_serde/parquet_timestamp.h"
+#include "format_v2/parquet/native_schema_desc.h"
+#include "format_v2/parquet/reader/native/decoder.h"
+#include "format_v2/parquet/reader/native/level_decoder.h"
+#include "format_v2/parquet/reader/native/page_reader.h"
+#include "io/fs/buffered_reader.h"
+#include "runtime/runtime_profile.h"
+#include "storage/cache/page_cache.h"
+#include "util/bit_util.h"
+#include "util/block_compression.h"
+#include "util/unaligned.h"
+
+namespace cctz {
+class time_zone;
+} // namespace cctz
+namespace doris {
+namespace io {
+class BufferedStreamReader;
+struct IOContext;
+} // namespace io
+} // namespace doris
+
+namespace doris::format::parquet::native {
+
+bool can_prepare_page_cache_payload(bool session_cache_enabled, bool
storage_cache_disabled,
+ bool cache_available, bool
header_available) {
+ return session_cache_enabled && !storage_cache_disabled && cache_available
&& header_available;
+}
+
+ParquetReaderCompat parquet_reader_compat(const std::string& created_by) {
+ if (created_by.empty()) {
+ return {};
+ }
+ const ::parquet::ApplicationVersion version(created_by);
+ return {.parquet_816_padding =
+
version.VersionLt(::parquet::ApplicationVersion::PARQUET_816_FIXED_VERSION()),
+ .data_page_v2_always_compressed = version.VersionLt(
+
::parquet::ApplicationVersion::PARQUET_CPP_10353_FIXED_VERSION())};
+}
+
+Status compute_column_chunk_range(const tparquet::ColumnMetaData& metadata,
size_t file_size,
+ bool parquet_816_padding, ColumnChunkRange*
range) {
+ DORIS_CHECK(range != nullptr);
+ int64_t start = metadata.data_page_offset;
+ if (metadata.__isset.dictionary_page_offset &&
metadata.dictionary_page_offset > 0 &&
+ metadata.dictionary_page_offset < start) {
+ // Some writers use dictionary_page_offset=0 as an absence sentinel.
Range validation must
+ // follow has_dict_page() or the native reader starts at the Parquet
magic bytes.
+ start = metadata.dictionary_page_offset;
+ }
+ const int64_t length = metadata.total_compressed_size;
+ if (UNLIKELY(start < 0 || length < 0)) {
+ return Status::Corruption("Parquet column chunk has a negative offset
or length");
+ }
+ const uint64_t unsigned_start = static_cast<uint64_t>(start);
+ const uint64_t unsigned_length = static_cast<uint64_t>(length);
+ if (UNLIKELY(unsigned_start > file_size || unsigned_length > file_size -
unsigned_start)) {
+ // Thrift range fields are signed and untrusted; validate before
converting them to the
+ // unsigned stream-reader coordinates so overflow cannot wrap back
into the file.
+ return Status::Corruption("Parquet column chunk [{}, {}) exceeds file
size {}", start,
+ unsigned_start + unsigned_length, file_size);
+ }
+ size_t bounded_length = static_cast<size_t>(unsigned_length);
+ if (parquet_816_padding) {
+ // parquet-mr before PARQUET-816 under-reported the chunk by up to 100
bytes. Padding stays
+ // file-bounded and is only enabled for the affected writer versions.
+ bounded_length += std::min<size_t>(100, file_size - unsigned_start -
unsigned_length);
+ }
+ range->offset = static_cast<size_t>(unsigned_start);
+ range->length = bounded_length;
+ return Status::OK();
+}
+
+bool validate_offset_index(const tparquet::OffsetIndex& index, const
ColumnChunkRange& chunk_range,
+ int64_t data_page_offset, int64_t row_count) {
+ if (index.page_locations.empty() || data_page_offset < 0 || row_count < 0
||
+ index.page_locations.front().first_row_index != 0 ||
+ index.page_locations.front().offset != data_page_offset ||
+ chunk_range.length > std::numeric_limits<size_t>::max() -
chunk_range.offset) {
+ return false;
+ }
+ // Row indexes alone cannot detect a uniformly shifted OffsetIndex. Anchor
its first location
+ // to the owning metadata so page-to-row mapping cannot silently move by
one physical page.
+ const uint64_t chunk_begin = chunk_range.offset;
+ const uint64_t chunk_end = chunk_begin + chunk_range.length;
+ uint64_t previous_end = chunk_begin;
+ int64_t previous_row = -1;
+ for (const auto& location : index.page_locations) {
+ if (location.first_row_index <= previous_row ||
location.first_row_index >= row_count ||
+ location.offset < 0 || location.compressed_page_size <= 0) {
+ return false;
+ }
+ const uint64_t begin = static_cast<uint64_t>(location.offset);
+ const uint64_t size =
static_cast<uint64_t>(location.compressed_page_size);
+ if (begin < chunk_begin || begin < previous_end || begin > chunk_end ||
+ size > chunk_end - begin) {
+ return false;
+ }
+ previous_row = location.first_row_index;
+ previous_end = begin + size;
+ }
+ return true;
+}
+
+namespace {
+
+Status append_v2_int96_datetime(ColumnDateTimeV2::Container& data,
+ const ParquetInt96Timestamp& value,
+ const cctz::time_zone& timezone) {
+ static constexpr int32_t JULIAN_EPOCH_OFFSET_DAYS = 2440588;
+ static constexpr int64_t MICROS_PER_DAY = 86400000000LL;
+ static constexpr int64_t MICROS_PER_SECOND = 1000000LL;
+
+ // Arrow normalized out-of-day INT96 nanos before the V2 native path
replaced it. Preserve that
+ // file-compatibility invariant here; rejecting the raw nanos loses
otherwise valid year-0 and
+ // pre-epoch timestamps used by existing Parquet files.
+ const __int128 days = static_cast<__int128>(value.julian_day) -
JULIAN_EPOCH_OFFSET_DAYS;
+ // The compatibility cast truncates the signed nanos field itself.
Normalizing it to a positive
+ // time-of-day first changes negative out-of-day values by one microsecond.
+ const __int128 timestamp_micros = days * MICROS_PER_DAY +
value.nanos_of_day / 1000;
+ if (timestamp_micros < std::numeric_limits<int64_t>::min() ||
+ timestamp_micros > std::numeric_limits<int64_t>::max()) {
+ return Status::DataQualityError("Parquet INT96 timestamp overflows
microseconds");
+ }
+
+ const int64_t micros = static_cast<int64_t>(timestamp_micros);
+ int64_t epoch_seconds = micros / MICROS_PER_SECOND;
+ int64_t micros_of_second = micros % MICROS_PER_SECOND;
+ if (micros_of_second < 0) {
+ micros_of_second += MICROS_PER_SECOND;
+ --epoch_seconds;
+ }
+ DateV2Value<DateTimeV2ValueType> datetime;
+ datetime.from_unixtime(epoch_seconds, timezone);
+ datetime.set_microsecond(static_cast<uint32_t>(micros_of_second));
+ if (!datetime.is_valid_date()) {
+ return Status::DataQualityError("Parquet INT96 timestamp is outside
the Doris range");
+ }
+ data.push_back(datetime);
+ return Status::OK();
+}
+
+class V2Int96DateTimeConsumer final : public ParquetFixedValueConsumer {
+public:
+ V2Int96DateTimeConsumer(IColumn& column, const ParquetDecodeContext&
context,
+ ParquetMaterializationState* state)
+ : _data(assert_cast<ColumnDateTimeV2&>(column).get_data()),
_state(state) {
+ static const auto utc = cctz::utc_time_zone();
+ _timezone = context.timezone == nullptr ? &utc : context.timezone;
+ }
+
+ Status consume(const uint8_t* values, size_t num_values, size_t
value_width) override {
+ DORIS_CHECK_EQ(value_width, sizeof(ParquetInt96Timestamp));
+ const size_t old_size = _data.size();
+ for (size_t row = 0; row < num_values; ++row) {
+ const auto value = unaligned_load<ParquetInt96Timestamp>(
+ values + row * sizeof(ParquetInt96Timestamp));
+ const auto status = append_v2_int96_datetime(_data, value,
*_timezone);
+ if (!status.ok()) {
+ if (_state != nullptr &&
_state->mark_conversion_failure(_data.size())) {
+ _data.emplace_back();
+ continue;
+ }
+ _data.resize(old_size);
+ return status;
+ }
+ }
+ return Status::OK();
+ }
+
+private:
+ ColumnDateTimeV2::Container& _data;
+ ParquetMaterializationState* _state;
+ const cctz::time_zone* _timezone = nullptr;
+};
+
+class RejectV2Int96BinaryConsumer final : public ParquetBinaryValueConsumer {
+public:
+ Status consume(const StringRef*, size_t) override {
+ return Status::NotSupported("INT96 cannot be decoded from binary
Parquet values");
+ }
+};
+
+Status read_v2_int96_datetime(IColumn& column, ParquetDecodeSource& source,
+ const ParquetDecodeContext& context, size_t
num_values,
+ ParquetMaterializationState& state) {
+ V2Int96DateTimeConsumer consumer(column, context, &state);
+ if (context.encoding != ParquetValueEncoding::DICTIONARY) {
+ return source.decode_fixed_values(num_values, consumer);
+ }
+ if (state.dictionary_generation != source.dictionary_generation()) {
+ state.typed_dictionary = column.clone_empty();
+ auto* output_null_map =
state.begin_dictionary_conversion(source.dictionary_size());
+ V2Int96DateTimeConsumer dictionary_consumer(*state.typed_dictionary,
context, &state);
+ RejectV2Int96BinaryConsumer binary_consumer;
+ const auto dictionary_status =
+ source.decode_dictionary(dictionary_consumer, binary_consumer);
+ state.end_dictionary_conversion(output_null_map);
+ RETURN_IF_ERROR(dictionary_status);
+ DORIS_CHECK_EQ(state.typed_dictionary->size(),
source.dictionary_size());
+ state.dictionary_generation = source.dictionary_generation();
+ }
+ RETURN_IF_ERROR(source.decode_dictionary_indices(num_values,
&state.dictionary_indices));
+ DORIS_CHECK_EQ(state.dictionary_indices.size(), num_values);
+ const size_t old_size = column.size();
+ column.insert_indices_from(*state.typed_dictionary,
state.dictionary_indices.data(),
+ state.dictionary_indices.data() + num_values);
+ if (state.can_insert_null_on_conversion_failure()) {
+ for (size_t row = 0; row < num_values; ++row) {
+ if (!state.dictionary_conversion_failures.empty() &&
+
state.dictionary_conversion_failures[state.dictionary_indices[row]] != 0) {
+ state.mark_conversion_failure(old_size + row);
+ }
+ }
+ }
+ return Status::OK();
+}
+
+Status read_native_or_serde(IColumn& column, const DataTypeSerDe& serde,
+ ParquetDecodeSource& source, const
ParquetDecodeContext& context,
+ size_t num_values, ParquetMaterializationState&
state) {
+ if (context.physical_type == ParquetPhysicalType::INT96 &&
+ check_and_get_column<ColumnDateTimeV2>(&column) != nullptr) {
+ return read_v2_int96_datetime(column, source, context, num_values,
state);
+ }
+ return serde.read_column_from_parquet(column, source, context, num_values,
state);
+}
+
+Status translate_value_encoding(tparquet::Encoding::type encoding,
+ ParquetValueEncoding* translated) {
+ DORIS_CHECK(translated != nullptr);
+ switch (encoding) {
+ case tparquet::Encoding::PLAIN:
+ *translated = ParquetValueEncoding::PLAIN;
+ return Status::OK();
+ case tparquet::Encoding::RLE_DICTIONARY:
+ case tparquet::Encoding::PLAIN_DICTIONARY:
+ *translated = ParquetValueEncoding::DICTIONARY;
+ return Status::OK();
+ case tparquet::Encoding::RLE:
+ *translated = ParquetValueEncoding::RLE;
+ return Status::OK();
+ case tparquet::Encoding::BIT_PACKED:
+ *translated = ParquetValueEncoding::BIT_PACKED;
+ return Status::OK();
+ case tparquet::Encoding::DELTA_BINARY_PACKED:
+ *translated = ParquetValueEncoding::DELTA_BINARY_PACKED;
+ return Status::OK();
+ case tparquet::Encoding::DELTA_LENGTH_BYTE_ARRAY:
+ *translated = ParquetValueEncoding::DELTA_LENGTH_BYTE_ARRAY;
+ return Status::OK();
+ case tparquet::Encoding::DELTA_BYTE_ARRAY:
+ *translated = ParquetValueEncoding::DELTA_BYTE_ARRAY;
+ return Status::OK();
+ case tparquet::Encoding::BYTE_STREAM_SPLIT:
+ *translated = ParquetValueEncoding::BYTE_STREAM_SPLIT;
+ return Status::OK();
+ default:
+ return Status::NotSupported("Unsupported Parquet encoding {}",
+ tparquet::to_string(encoding));
+ }
+}
+
+template <bool HAS_FILTER>
+Status decode_selected_values(IColumn& column, const DataTypeSerDe& serde,
Decoder& decoder,
+ const ParquetDecodeContext& context,
+ ParquetMaterializationState& state,
ColumnSelectVector& select_vector,
+ int64_t* materialization_time) {
+ SCOPED_RAW_TIMER(materialization_time);
+ ColumnSelectVector::DataReadType read_type;
+ while (const size_t run_length =
select_vector.get_next_run<HAS_FILTER>(&read_type)) {
+ switch (read_type) {
+ case ColumnSelectVector::CONTENT:
+ RETURN_IF_ERROR(
+ read_native_or_serde(column, serde, decoder, context,
run_length, state));
+ break;
+ case ColumnSelectVector::NULL_DATA:
+ column.insert_many_defaults(run_length);
+ break;
+ case ColumnSelectVector::FILTERED_CONTENT:
+ RETURN_IF_ERROR(decoder.skip_values(run_length));
+ break;
+ case ColumnSelectVector::FILTERED_NULL:
+ break;
+ }
+ }
+ return Status::OK();
+}
+
+// Presents one sparse page request as an ordinary sequential source to
DataTypeSerDe. SerDe is
+// entered once per page fragment; the concrete decoder decides whether to
gather selected spans,
+// batch-decode and compact, or use the cursor-preserving range fallback.
+class SelectedDecodeSource final : public ParquetDecodeSource {
+public:
+ SelectedDecodeSource(Decoder& decoder, const ParquetSelection& selection)
+ : _decoder(decoder), _selection(selection) {}
+
+ Status decode_fixed_values(size_t num_values, ParquetFixedValueConsumer&
consumer) override {
+ DORIS_CHECK_EQ(num_values, _selection.selected_values);
+ return _decoder.decode_selected_fixed_values(_selection, consumer);
+ }
+
+ Status decode_binary_values(size_t num_values, ParquetBinaryValueConsumer&
consumer) override {
+ DORIS_CHECK_EQ(num_values, _selection.selected_values);
+ return _decoder.decode_selected_binary_values(_selection, consumer);
+ }
+
+ Status skip_values(size_t num_values) override {
+ return Status::NotSupported("Selected Parquet source cannot be
skipped, values={}",
+ num_values);
+ }
+
+ bool has_dictionary() const override { return _decoder.has_dictionary(); }
+ uint64_t dictionary_generation() const override { return
_decoder.dictionary_generation(); }
+ size_t dictionary_size() const override { return
_decoder.dictionary_size(); }
+
+ Status decode_dictionary(ParquetFixedValueConsumer& fixed_consumer,
+ ParquetBinaryValueConsumer& binary_consumer)
override {
+ return _decoder.decode_dictionary(fixed_consumer, binary_consumer);
+ }
+
+ Status decode_dictionary_indices(size_t num_values, std::vector<uint32_t>*
indices) override {
+ DORIS_CHECK_EQ(num_values, _selection.selected_values);
+ return _decoder.decode_selected_dictionary_indices(_selection,
indices);
+ }
+
+private:
+ Decoder& _decoder;
+ const ParquetSelection& _selection;
+};
+
+Status decode_selected_non_null_values(IColumn& column, const DataTypeSerDe&
serde,
+ Decoder& decoder, const
ParquetDecodeContext& context,
+ ParquetMaterializationState& state,
+ ColumnSelectVector& select_vector,
+ int64_t* materialization_time) {
+ auto& selection = state.selection;
+ selection.ranges.clear();
+ selection.total_values = select_vector.num_values();
+ selection.selected_values = 0;
+
+ size_t cursor = 0;
+ ColumnSelectVector::DataReadType read_type;
+ while (const size_t run_length =
select_vector.get_next_run<true>(&read_type)) {
+ DORIS_CHECK(read_type == ColumnSelectVector::CONTENT ||
+ read_type == ColumnSelectVector::FILTERED_CONTENT);
+ if (read_type == ColumnSelectVector::CONTENT) {
+ selection.ranges.push_back({.first = cursor, .count = run_length});
+ selection.selected_values += run_length;
+ }
+ cursor += run_length;
+ }
+ DORIS_CHECK_EQ(cursor, selection.total_values);
+ if (selection.selected_values == 0) {
+ return decoder.skip_values(selection.total_values);
+ }
+
+ SCOPED_RAW_TIMER(materialization_time);
+ SelectedDecodeSource selected_source(decoder, selection);
+ return read_native_or_serde(column, serde, selected_source, context,
selection.selected_values,
+ state);
+}
+
+} // namespace
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::ColumnChunkReader(
+ io::BufferedStreamReader* reader, tparquet::ColumnChunk* column_chunk,
+ NativeFieldSchema* field_schema, const tparquet::OffsetIndex*
offset_index,
+ size_t total_rows, io::IOContext* io_ctx, const
ParquetPageReadContext& page_read_ctx,
+ const ColumnChunkRange* chunk_range)
+ : _field_schema(field_schema),
+ _max_rep_level(field_schema->repetition_level),
+ _max_def_level(field_schema->definition_level),
+ _stream_reader(reader),
+ _metadata(column_chunk->meta_data),
+ _offset_index(offset_index),
+ _total_rows(total_rows),
+ _io_ctx(io_ctx),
+ _page_read_ctx(page_read_ctx) {
+ if (chunk_range != nullptr) {
+ _chunk_range = *chunk_range;
+ _has_validated_chunk_range = true;
+ }
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::init() {
+ size_t start_offset = _has_validated_chunk_range
+ ? _chunk_range.offset
+ : (has_dict_page(_metadata) ?
_metadata.dictionary_page_offset
+ :
_metadata.data_page_offset);
+ size_t chunk_size =
+ _has_validated_chunk_range ? _chunk_range.length :
_metadata.total_compressed_size;
+ // create page reader
+ _page_reader = create_page_reader<IN_COLLECTION, OFFSET_INDEX>(
+ _stream_reader, _io_ctx, start_offset, chunk_size, _total_rows,
_metadata,
+ _page_read_ctx, _offset_index);
+ // get the block compression codec
+ RETURN_IF_ERROR(get_block_compression_codec(_metadata.codec,
&_block_compress_codec));
+ _state = INITIALIZED;
+ RETURN_IF_ERROR(_parse_first_page_header());
+ return Status::OK();
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::skip_nested_values(
+ const std::vector<level_t>& def_levels, size_t start_index) {
+ size_t no_value_cnt = 0;
+ size_t value_cnt = 0;
+
+ DORIS_CHECK(start_index <= def_levels.size());
+ for (size_t idx = start_index; idx < def_levels.size(); idx++) {
+ level_t def_level = def_levels[idx];
+ if (IN_COLLECTION && def_level <
_field_schema->repeated_parent_def_level) {
+ no_value_cnt++;
+ } else if (def_level < _field_schema->definition_level) {
+ no_value_cnt++;
+ } else {
+ value_cnt++;
+ }
+ }
+
+ RETURN_IF_ERROR(skip_values(value_cnt, true));
+ RETURN_IF_ERROR(skip_values(no_value_cnt, false));
+ return Status::OK();
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::read_levels(
+ size_t num_values, std::vector<level_t>* rep_levels,
std::vector<level_t>* def_levels) {
+ DORIS_CHECK(rep_levels != nullptr);
+ DORIS_CHECK(def_levels != nullptr);
+ if (_remaining_num_values < num_values || _remaining_rep_nums < num_values
||
+ _remaining_def_nums < num_values) {
+ return Status::Corruption(
+ "Parquet level reader requested {} slots with only {}/{}/{}
remaining", num_values,
+ _remaining_num_values, _remaining_rep_nums,
_remaining_def_nums);
+ }
+
+ const size_t start_index = def_levels->size();
+ rep_levels->resize(rep_levels->size() + num_values, 0);
+ def_levels->resize(def_levels->size() + num_values, 0);
+ if (_max_rep_level > 0) {
+ const size_t decoded = _rep_level_decoder.get_levels(
+ rep_levels->data() + rep_levels->size() - num_values,
num_values);
+ if (decoded != num_values) {
+ return Status::Corruption("Parquet repetition level stream ended
after {} of {} slots",
+ decoded, num_values);
+ }
+ }
+ if (_max_def_level > 0) {
+ const size_t decoded = _def_level_decoder.get_levels(
+ def_levels->data() + def_levels->size() - num_values,
num_values);
+ if (decoded != num_values) {
+ return Status::Corruption("Parquet definition level stream ended
after {} of {} slots",
+ decoded, num_values);
+ }
+ }
+ _remaining_rep_nums -= num_values;
+ _remaining_def_nums -= num_values;
+ return skip_nested_values(*def_levels, start_index);
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+Status ColumnChunkReader<IN_COLLECTION,
OFFSET_INDEX>::_parse_first_page_header() {
+ while (true) {
+ RETURN_IF_ERROR(_page_reader->parse_page_header());
+ const tparquet::PageHeader* header = nullptr;
+ RETURN_IF_ERROR(_page_reader->get_page_header(&header));
+ if (header->type == tparquet::PageType::DATA_PAGE ||
+ header->type == tparquet::PageType::DATA_PAGE_V2) {
+ _state = INITIALIZED;
+ return parse_page_header();
+ }
+ if (header->type != tparquet::PageType::DICTIONARY_PAGE) {
+ RETURN_IF_ERROR(_page_reader->skip_auxiliary_page());
+ _state = INITIALIZED;
+ continue;
+ }
+ // the first page maybe directory page even if
_metadata.__isset.dictionary_page_offset == false,
+ // so we should parse the directory page in next_page()
+ RETURN_IF_ERROR(_decode_dict_page());
+ // parse the real first data page
+ RETURN_IF_ERROR(_page_reader->dict_next_page());
+ _state = INITIALIZED;
+ // A dictionary is the only non-data page with decoder state. Any
following index or
+ // extension pages are skipped by the same pre-data loop.
+ }
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::parse_page_header() {
+ if (_state == HEADER_PARSED || _state == DATA_LOADED) {
+ return Status::OK();
+ }
+ const tparquet::PageHeader* header = nullptr;
+ while (true) {
+ RETURN_IF_ERROR(_page_reader->parse_page_header());
+ RETURN_IF_ERROR(_page_reader->get_page_header(&header));
+ if (header->type == tparquet::PageType::DATA_PAGE ||
+ header->type == tparquet::PageType::DATA_PAGE_V2) {
+ break;
+ }
+ if (header->type == tparquet::PageType::DICTIONARY_PAGE) {
+ return Status::Corruption("Parquet dictionary page appears after
data pages");
+ }
+ RETURN_IF_ERROR(_page_reader->skip_auxiliary_page());
+ }
+ int32_t page_num_values = _page_reader->is_header_v2() ?
header->data_page_header_v2.num_values
+ :
header->data_page_header.num_values;
+ if (page_num_values < 0 || page_num_values > _metadata.num_values ||
+ (!OFFSET_INDEX &&
+ static_cast<uint64_t>(page_num_values) >
+ static_cast<uint64_t>(_metadata.num_values) -
_chunk_parsed_values)) {
+ // Page counts are untrusted and feed both level decoders and scratch
sizing. Bound each
+ // page by the column metadata before converting to unsigned counters.
+ return Status::Corruption("Parquet data page value count {} exceeds
column total {}",
+ page_num_values, _metadata.num_values);
+ }
+ if constexpr (!IN_COLLECTION) {
+ const size_t page_start_row = _page_reader->start_row();
+ const size_t page_end_row = _page_reader->end_row();
+ if (UNLIKELY(page_end_row < page_start_row ||
+ static_cast<size_t>(page_num_values) != page_end_row -
page_start_row)) {
+ // Flat columns have exactly one physical value slot per logical
row. Rejecting a
+ // divergent header/OffsetIndex span prevents every later page
from shifting rows.
+ return Status::Corruption(
+ "Parquet flat data page has {} values for logical row
range [{}, {})",
+ page_num_values, page_start_row, page_end_row);
+ }
+ }
+ _remaining_rep_nums = page_num_values;
+ _remaining_def_nums = page_num_values;
+ _remaining_num_values = page_num_values;
+
+ // no offset will parse all header.
+ if constexpr (OFFSET_INDEX == false) {
+ _chunk_parsed_values += _remaining_num_values;
+ }
+ _state = HEADER_PARSED;
+ return Status::OK();
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::next_page() {
+ _state = INITIALIZED;
+ RETURN_IF_ERROR(_page_reader->next_page());
+ return Status::OK();
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+Status ColumnChunkReader<IN_COLLECTION,
OFFSET_INDEX>::_get_uncompressed_levels(
+ const tparquet::DataPageHeaderV2& page_v2, Slice& page_data) {
+ const size_t rl = page_v2.repetition_levels_byte_length;
+ const size_t dl = page_v2.definition_levels_byte_length;
+ if (UNLIKELY(rl > page_data.size || dl > page_data.size - rl)) {
+ // Validate the physical slice again because a cached entry may itself
be truncated.
+ return Status::Corruption("Parquet data page v2 level bytes exceed
available payload");
+ }
+ _v2_rep_levels = Slice(page_data.data, rl);
+ _v2_def_levels = Slice(page_data.data + rl, dl);
+ page_data.data += dl + rl;
+ page_data.size -= dl + rl;
+ return Status::OK();
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::load_page_data() {
+ if (_state == DATA_LOADED) {
+ return Status::OK();
+ }
+ if (UNLIKELY(_state != HEADER_PARSED)) {
+ return Status::Corruption("Should parse page header");
+ }
+
+ const tparquet::PageHeader* header = nullptr;
+ RETURN_IF_ERROR(_page_reader->get_page_header(&header));
+ int32_t uncompressed_size = header->uncompressed_page_size;
+ bool page_loaded = false;
+
+ // First, try to reuse a cache handle previously discovered by PageReader
+ // (header-only lookup) to avoid a second lookup here.
+ if (_page_read_ctx.enable_parquet_file_page_cache &&
!config::disable_storage_page_cache &&
+ StoragePageCache::instance() != nullptr) {
+ if (_page_reader->has_page_cache_handle()) {
+ const PageCacheHandle& handle = _page_reader->page_cache_handle();
+ Slice cached = handle.data();
+ size_t header_size = _page_reader->header_bytes().size();
+ size_t levels_size = 0;
+ if (header->__isset.data_page_header_v2) {
+ const tparquet::DataPageHeaderV2& header_v2 =
header->data_page_header_v2;
+ size_t rl = header_v2.repetition_levels_byte_length;
+ size_t dl = header_v2.definition_levels_byte_length;
+ levels_size = rl + dl;
+ if (UNLIKELY(header_size > cached.size ||
+ levels_size > cached.size - header_size)) {
+ return Status::Corruption("Cached Parquet page is shorter
than its v2 levels");
+ }
+ _v2_rep_levels =
+ Slice(reinterpret_cast<const uint8_t*>(cached.data) +
header_size, rl);
+ _v2_def_levels =
+ Slice(reinterpret_cast<const uint8_t*>(cached.data) +
header_size + rl, dl);
+ }
+ // payload_slice points to the bytes after header and levels
+ if (UNLIKELY(header_size + levels_size > cached.size)) {
+ return Status::Corruption("Cached Parquet page is shorter than
its header");
+ }
+ Slice payload_slice(cached.data + header_size + levels_size,
+ cached.size - header_size - levels_size);
+
+ bool cache_payload_is_decompressed =
_page_reader->is_cache_payload_decompressed();
+ const size_t expected_payload_size =
+ cache_payload_is_decompressed
+ ?
static_cast<size_t>(header->uncompressed_page_size) - levels_size
+ :
static_cast<size_t>(header->compressed_page_size) - levels_size;
+ if (UNLIKELY(payload_slice.size != expected_payload_size)) {
+ return Status::Corruption("Cached Parquet page payload has
size {}, expected {}",
+ payload_slice.size,
expected_payload_size);
+ }
+
+ if (cache_payload_is_decompressed) {
+ // Cached payload is already uncompressed
+ _page_data = payload_slice;
+ } else {
+ CHECK(_block_compress_codec);
+ // Decompress cached payload into _decompress_buf for decoding
+ size_t uncompressed_payload_size =
+ header->__isset.data_page_header_v2
+ ?
static_cast<size_t>(header->uncompressed_page_size) - levels_size
+ :
static_cast<size_t>(header->uncompressed_page_size);
+ _reserve_decompress_buf(uncompressed_payload_size);
+ _page_data = Slice(_decompress_buf.get(),
uncompressed_payload_size);
+ SCOPED_RAW_TIMER(&_chunk_statistics.decompress_time);
+ _chunk_statistics.decompress_cnt++;
+
RETURN_IF_ERROR(_block_compress_codec->decompress(payload_slice, &_page_data));
+ if (UNLIKELY(_page_data.size != uncompressed_payload_size)) {
+ return Status::Corruption("Parquet page decompressed to {}
bytes, expected {}",
+ _page_data.size,
uncompressed_payload_size);
+ }
+ }
+ // page cache counters were incremented when PageReader did the
header-only
+ // cache lookup. Do not increment again to avoid double-counting.
+ page_loaded = true;
+ }
+ }
+
+ if (!page_loaded) {
+ const bool prepare_cache_payload = can_prepare_page_cache_payload(
+ _page_read_ctx.enable_parquet_file_page_cache,
config::disable_storage_page_cache,
+ StoragePageCache::instance() != nullptr,
!_page_reader->header_bytes().empty());
+ if (_block_compress_codec != nullptr) {
+ Slice compressed_data;
+ RETURN_IF_ERROR(_page_reader->get_page_data(compressed_data));
+ std::vector<uint8_t> level_bytes;
+ if (header->__isset.data_page_header_v2) {
+ const tparquet::DataPageHeaderV2& header_v2 =
header->data_page_header_v2;
+ // uncompressed_size = rl + dl + uncompressed_data_size
+ // compressed_size = rl + dl + compressed_data_size
+ uncompressed_size -= header_v2.repetition_levels_byte_length +
+ header_v2.definition_levels_byte_length;
+ // copy level bytes (rl + dl) so that we can cache header +
levels + uncompressed payload
+ size_t rl = header_v2.repetition_levels_byte_length;
+ size_t dl = header_v2.definition_levels_byte_length;
+ size_t level_sz = rl + dl;
+ if (prepare_cache_payload && level_sz > 0) {
+ level_bytes.resize(level_sz);
+ memcpy(level_bytes.data(), compressed_data.data, level_sz);
+ }
+ // now remove levels from compressed_data for decompression
+ RETURN_IF_ERROR(_get_uncompressed_levels(header_v2,
compressed_data));
+ }
+ bool is_v2_compressed = header->__isset.data_page_header_v2 &&
+ (header->data_page_header_v2.is_compressed
||
+
_page_read_ctx.data_page_v2_always_compressed);
+ bool page_has_compression = header->__isset.data_page_header ||
is_v2_compressed;
+
+ if (page_has_compression) {
+ // Decompress payload for immediate decoding
+ _reserve_decompress_buf(uncompressed_size);
+ _page_data = Slice(_decompress_buf.get(), uncompressed_size);
+ SCOPED_RAW_TIMER(&_chunk_statistics.decompress_time);
+ _chunk_statistics.decompress_cnt++;
+
RETURN_IF_ERROR(_block_compress_codec->decompress(compressed_data,
&_page_data));
+ if (UNLIKELY(_page_data.size !=
static_cast<size_t>(uncompressed_size))) {
+ return Status::Corruption("Parquet page decompressed to {}
bytes, expected {}",
+ _page_data.size,
uncompressed_size);
+ }
+
+ // Decide whether to cache decompressed payload or compressed
payload based on threshold
+ bool cache_payload_decompressed = should_cache_decompressed(
+ header, _metadata,
_page_read_ctx.data_page_v2_always_compressed);
+
+ if (prepare_cache_payload) {
+ if (cache_payload_decompressed) {
+ _insert_page_into_cache(level_bytes, _page_data);
+
_chunk_statistics.page_cache_decompressed_write_counter += 1;
+ } else {
+ if (config::enable_parquet_cache_compressed_pages) {
+ // cache the compressed payload as-is (header |
levels | compressed_payload)
+ _insert_page_into_cache(
+ level_bytes, Slice(compressed_data.data,
compressed_data.size));
+
_chunk_statistics.page_cache_compressed_write_counter += 1;
+ }
+ }
+ }
+ } else {
+ // no compression on this page, use the data directly
+ _page_data = Slice(compressed_data.data, compressed_data.size);
+ if (prepare_cache_payload) {
+ _insert_page_into_cache(level_bytes, _page_data);
+ _chunk_statistics.page_cache_decompressed_write_counter +=
1;
+ }
+ }
+ } else {
+ // For uncompressed page, we may still need to extract v2 levels
+ std::vector<uint8_t> level_bytes;
+ Slice uncompressed_data;
+ RETURN_IF_ERROR(_page_reader->get_page_data(uncompressed_data));
+ if (header->__isset.data_page_header_v2) {
+ const tparquet::DataPageHeaderV2& header_v2 =
header->data_page_header_v2;
+ size_t rl = header_v2.repetition_levels_byte_length;
+ size_t dl = header_v2.definition_levels_byte_length;
+ size_t level_sz = rl + dl;
+ if (prepare_cache_payload && level_sz > 0) {
+ level_bytes.resize(level_sz);
+ memcpy(level_bytes.data(), uncompressed_data.data,
level_sz);
+ }
+ RETURN_IF_ERROR(_get_uncompressed_levels(header_v2,
uncompressed_data));
+ }
+ // copy page data out
+ _page_data = Slice(uncompressed_data.data, uncompressed_data.size);
+ // Optionally cache uncompressed data for uncompressed pages
+ if (prepare_cache_payload) {
+ _insert_page_into_cache(level_bytes, _page_data);
+ _chunk_statistics.page_cache_decompressed_write_counter += 1;
+ }
+ }
+ }
+
+ // Initialize repetition level and definition level. Skip when level = 0,
which means required field.
+ if (_max_rep_level > 0) {
+ SCOPED_RAW_TIMER(&_chunk_statistics.decode_level_time);
+ if (header->__isset.data_page_header_v2) {
+ RETURN_IF_ERROR(_rep_level_decoder.init_v2(_v2_rep_levels,
_max_rep_level,
+ _remaining_rep_nums));
+ } else {
+ RETURN_IF_ERROR(_rep_level_decoder.init(
+ &_page_data,
header->data_page_header.repetition_level_encoding, _max_rep_level,
+ _remaining_rep_nums));
+ }
+ }
+ if (_max_def_level > 0) {
+ SCOPED_RAW_TIMER(&_chunk_statistics.decode_level_time);
+ if (header->__isset.data_page_header_v2) {
+ RETURN_IF_ERROR(_def_level_decoder.init_v2(_v2_def_levels,
_max_def_level,
+ _remaining_def_nums));
+ } else {
+ RETURN_IF_ERROR(_def_level_decoder.init(
+ &_page_data,
header->data_page_header.definition_level_encoding, _max_def_level,
+ _remaining_def_nums));
+ }
+ }
+ auto encoding = header->__isset.data_page_header_v2 ?
header->data_page_header_v2.encoding
+ :
header->data_page_header.encoding;
+ // change the deprecated encoding to RLE_DICTIONARY
+ if (encoding == tparquet::Encoding::PLAIN_DICTIONARY) {
+ encoding = tparquet::Encoding::RLE_DICTIONARY;
+ }
+ _current_encoding = encoding;
+
+ // Reuse page decoder
+ if (_decoders.find(static_cast<int>(encoding)) != _decoders.end()) {
+ _page_decoder = _decoders[static_cast<int>(encoding)].get();
+ } else {
+ std::unique_ptr<Decoder> page_decoder;
+ RETURN_IF_ERROR(Decoder::get_decoder(_metadata.type, encoding,
page_decoder));
+ // Set type length
+ page_decoder->set_type_length(_get_type_length());
+ _decoders[static_cast<int>(encoding)] = std::move(page_decoder);
+ _page_decoder = _decoders[static_cast<int>(encoding)].get();
+ }
+ // Encoding headers cannot legitimately advertise more physical values
than the data page's
+ // logical value count; establish the bound before decoders inspect
external counts.
+ _page_decoder->set_expected_values(_remaining_num_values);
+ RETURN_IF_ERROR(_page_decoder->set_data(&_page_data));
+
+ _state = DATA_LOADED;
+ return Status::OK();
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::_decode_dict_page() {
+ const tparquet::PageHeader* header = nullptr;
+ RETURN_IF_ERROR(_page_reader->get_page_header(&header));
+ DCHECK_EQ(tparquet::PageType::DICTIONARY_PAGE, header->type);
+ SCOPED_RAW_TIMER(&_chunk_statistics.decode_dict_time);
+
+ // Using the PLAIN_DICTIONARY enum value is deprecated in the Parquet 2.0
specification.
+ // Prefer using RLE_DICTIONARY in a data page and PLAIN in a dictionary
page for Parquet 2.0+ files.
+ // refer: https://github.com/apache/parquet-format/blob/master/Encodings.md
+ tparquet::Encoding::type dict_encoding =
header->dictionary_page_header.encoding;
+ if (dict_encoding != tparquet::Encoding::PLAIN_DICTIONARY &&
+ dict_encoding != tparquet::Encoding::PLAIN) {
+ return Status::InternalError("Unsupported dictionary encoding {}",
+ tparquet::to_string(dict_encoding));
+ }
+
+ // Prepare dictionary data
+ int32_t uncompressed_size = header->uncompressed_page_size;
+ if (_block_compress_codec == nullptr &&
+ UNLIKELY(header->compressed_page_size != uncompressed_size)) {
+ // UNCOMPRESSED pages use the compressed size as their physical copy
length.
+ return Status::Corruption(
+ "Uncompressed Parquet dictionary sizes differ: compressed={},
uncompressed={}",
+ header->compressed_page_size, uncompressed_size);
+ }
+ auto dict_data = make_unique_buffer<uint8_t>(uncompressed_size);
+ bool dict_loaded = false;
+
+ // Try to load dictionary page from cache
+ if (_page_read_ctx.enable_parquet_file_page_cache &&
!config::disable_storage_page_cache &&
+ StoragePageCache::instance() != nullptr) {
+ if (_page_reader->has_page_cache_handle()) {
+ const PageCacheHandle& handle = _page_reader->page_cache_handle();
+ Slice cached = handle.data();
+ size_t header_size = _page_reader->header_bytes().size();
+ // Dictionary page layout in cache: header | payload (compressed
or uncompressed)
+ Slice payload_slice(cached.data + header_size, cached.size -
header_size);
+
+ bool cache_payload_is_decompressed =
_page_reader->is_cache_payload_decompressed();
+
+ if (cache_payload_is_decompressed) {
+ // Use cached decompressed dictionary data
+ if (UNLIKELY(payload_slice.size !=
static_cast<size_t>(uncompressed_size))) {
+ return Status::Corruption(
+ "Cached Parquet dictionary payload has size {},
expected {}",
+ payload_slice.size, uncompressed_size);
+ }
+ memcpy(dict_data.get(), payload_slice.data,
payload_slice.size);
+ dict_loaded = true;
+ } else {
+ CHECK(_block_compress_codec);
+ // Decompress cached compressed dictionary data
+ Slice dict_slice(dict_data.get(), uncompressed_size);
+
RETURN_IF_ERROR(_block_compress_codec->decompress(payload_slice, &dict_slice));
+ if (UNLIKELY(dict_slice.size !=
static_cast<size_t>(uncompressed_size))) {
+ return Status::Corruption(
+ "Parquet dictionary decompressed to {} bytes,
expected {}",
+ dict_slice.size, uncompressed_size);
+ }
+ dict_loaded = true;
+ }
+
+ // When dictionary page is loaded from cache, we need to skip the
page data
+ // to update the offset correctly (similar to calling
get_page_data())
+ if (dict_loaded) {
+ _page_reader->skip_page_data();
+ }
+ }
+ }
+
+ if (!dict_loaded) {
+ // Load and decompress dictionary page from file
+ if (_block_compress_codec != nullptr) {
+ auto dict_num = header->dictionary_page_header.num_values;
+ if (dict_num == 0 && uncompressed_size != 0) {
+ return Status::IOError(
+ "Dictionary page's num_values is {} but
uncompressed_size is {}", dict_num,
+ uncompressed_size);
+ }
+ Slice compressed_data;
+ Slice dict_slice(dict_data.get(), uncompressed_size);
+ if (dict_num != 0) {
+ RETURN_IF_ERROR(_page_reader->get_page_data(compressed_data));
+
RETURN_IF_ERROR(_block_compress_codec->decompress(compressed_data,
&dict_slice));
+ if (UNLIKELY(dict_slice.size !=
static_cast<size_t>(uncompressed_size))) {
+ return Status::Corruption(
+ "Parquet dictionary decompressed to {} bytes,
expected {}",
+ dict_slice.size, uncompressed_size);
+ }
+ }
+
+ // Decide whether to cache decompressed or compressed dictionary
based on threshold
+ // If uncompressed_page_size == 0, should_cache_decompressed will
return true
+ bool cache_payload_decompressed = should_cache_decompressed(
+ header, _metadata,
_page_read_ctx.data_page_v2_always_compressed);
+
+ if (_page_read_ctx.enable_parquet_file_page_cache &&
+ !config::disable_storage_page_cache &&
StoragePageCache::instance() != nullptr &&
+ !_page_reader->header_bytes().empty()) {
+ std::vector<uint8_t> empty_levels; // Dictionary pages don't
have levels
+ if (cache_payload_decompressed) {
+ // Cache the decompressed dictionary page
+ // If dict_num == 0, `dict_slice` will be empty
+ _insert_page_into_cache(empty_levels, dict_slice);
+ _chunk_statistics.page_cache_decompressed_write_counter +=
1;
+ } else {
+ if (config::enable_parquet_cache_compressed_pages) {
+ DCHECK(!compressed_data.empty());
+ // Cache the compressed dictionary page
+ _insert_page_into_cache(empty_levels,
+ Slice(compressed_data.data,
compressed_data.size));
+ _chunk_statistics.page_cache_compressed_write_counter
+= 1;
+ }
+ }
+ }
+ // `get_page_data` not called, we should skip the page data
+ // Because `_insert_page_into_cache` will use _page_reader, we
should exec `skip_page_data` after `_insert_page_into_cache`
+ if (dict_num == 0) {
+ _page_reader->skip_page_data();
+ }
+ } else {
+ Slice dict_slice;
+ RETURN_IF_ERROR(_page_reader->get_page_data(dict_slice));
+ // The data is stored by BufferedStreamReader, we should copy it
out
+ memcpy(dict_data.get(), dict_slice.data, dict_slice.size);
+
+ // Cache the uncompressed dictionary page
+ if (_page_read_ctx.enable_parquet_file_page_cache &&
+ !config::disable_storage_page_cache &&
StoragePageCache::instance() != nullptr &&
+ !_page_reader->header_bytes().empty()) {
+ std::vector<uint8_t> empty_levels;
+ Slice payload(dict_data.get(), uncompressed_size);
+ _insert_page_into_cache(empty_levels, payload);
+ _chunk_statistics.page_cache_decompressed_write_counter += 1;
+ }
+ }
+ }
+
+ // Cache page decoder
+ std::unique_ptr<Decoder> page_decoder;
+ RETURN_IF_ERROR(
+ Decoder::get_decoder(_metadata.type,
tparquet::Encoding::RLE_DICTIONARY, page_decoder));
+ // Set type length
+ page_decoder->set_type_length(_get_type_length());
+ // Set the dictionary data
+ RETURN_IF_ERROR(page_decoder->set_dict(dict_data, uncompressed_size,
+
header->dictionary_page_header.num_values));
+ _decoders[static_cast<int>(tparquet::Encoding::RLE_DICTIONARY)] =
std::move(page_decoder);
+
+ _has_dict = true;
+ return Status::OK();
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+void ColumnChunkReader<IN_COLLECTION,
OFFSET_INDEX>::_reserve_decompress_buf(size_t size) {
+ if (size > _decompress_buf_size) {
+ _decompress_buf_size = BitUtil::next_power_of_two(size);
+ _decompress_buf = make_unique_buffer<uint8_t>(_decompress_buf_size);
+ }
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+void ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::_insert_page_into_cache(
+ const std::vector<uint8_t>& level_bytes, const Slice& payload) {
+ StoragePageCache::CacheKey key =
+
_page_reader->make_page_cache_key(_page_reader->header_start_offset());
+ const std::vector<uint8_t>& header_bytes = _page_reader->header_bytes();
+ size_t total = header_bytes.size() + level_bytes.size() + payload.size;
+ auto page = std::make_unique<DataPage>(total, true, segment_v2::DATA_PAGE);
+ size_t pos = 0;
+ memcpy(page->data() + pos, header_bytes.data(), header_bytes.size());
+ pos += header_bytes.size();
+ if (!level_bytes.empty()) {
+ memcpy(page->data() + pos, level_bytes.data(), level_bytes.size());
+ pos += level_bytes.size();
+ }
+ if (payload.size > 0) {
+ memcpy(page->data() + pos, payload.data, payload.size);
+ pos += payload.size;
+ }
+ page->reset_size(total);
+ PageCacheHandle handle;
+ StoragePageCache::instance()->insert(key, page.get(), &handle,
segment_v2::DATA_PAGE);
+ page.release();
+ _chunk_statistics.page_cache_write_counter += 1;
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::skip_values(size_t
num_values,
+ bool
skip_data) {
+ if (UNLIKELY(_remaining_num_values < num_values)) {
+ return Status::IOError("Skip too many values in current page. {} vs.
{}",
+ _remaining_num_values, num_values);
+ }
+ if (skip_data) {
+ SCOPED_RAW_TIMER(&_chunk_statistics.decode_value_time);
+ RETURN_IF_ERROR(_page_decoder->skip_values(num_values));
+ }
+ // Commit logical page progress only after the physical decoder accepted
the whole request.
+ _remaining_num_values -= num_values;
+ return Status::OK();
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::materialize_values(
+ MutableColumnPtr& doris_column, const DataTypeSerDe& serde,
ParquetDecodeContext& context,
+ ParquetMaterializationState& state, ColumnSelectVector& select_vector)
{
+ if (select_vector.num_values() == 0) {
+ return Status::OK();
+ }
+ SCOPED_RAW_TIMER(&_chunk_statistics.decode_value_time);
+ if (UNLIKELY((doris_column->is_column_dictionary() ||
context.dictionary_index_only) &&
+ !_has_dict)) {
+ return Status::IOError("Not dictionary coded");
+ }
+ if (UNLIKELY(_remaining_num_values < select_vector.num_values())) {
+ return Status::IOError("Decode too many values in current page");
+ }
+ RETURN_IF_ERROR(translate_value_encoding(_current_encoding,
&context.encoding));
+ Status status;
+ if (select_vector.has_filter()) {
+ if (select_vector.num_nulls() == 0) {
+ ++_chunk_statistics.hybrid_selection_batches;
+ status = decode_selected_non_null_values(*doris_column, serde,
*_page_decoder, context,
+ state, select_vector,
+
&_chunk_statistics.materialization_time);
+ _chunk_statistics.hybrid_selection_ranges +=
state.selection.ranges.size();
+ } else {
+ ++_chunk_statistics.hybrid_selection_null_fallback_batches;
+ status = decode_selected_values<true>(*doris_column, serde,
*_page_decoder, context,
+ state, select_vector,
+
&_chunk_statistics.materialization_time);
+ }
+ } else {
+ status = decode_selected_values<false>(*doris_column, serde,
*_page_decoder, context, state,
+ select_vector,
+
&_chunk_statistics.materialization_time);
+ }
+ RETURN_IF_ERROR(status);
+ _remaining_num_values -= select_vector.num_values();
+ return Status::OK();
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+Status ColumnChunkReader<IN_COLLECTION,
OFFSET_INDEX>::seek_to_nested_row(size_t left_row) {
+ if constexpr (OFFSET_INDEX) {
+ while (true) {
+ if (_page_reader->start_row() <= left_row && left_row <
_page_reader->end_row()) {
+ break;
+ } else if (has_next_page()) {
+ RETURN_IF_ERROR(next_page());
+ _current_row = _page_reader->start_row();
+ } else [[unlikely]] {
+ return Status::InternalError("no match seek row {}, current
row {}", left_row,
+ _current_row);
+ }
+ };
+
+ RETURN_IF_ERROR(parse_page_header());
+ RETURN_IF_ERROR(load_page_data());
+ RETURN_IF_ERROR(_skip_nested_rows_in_page(left_row - _current_row));
+ _current_row = left_row;
+ } else {
+ while (true) {
+ RETURN_IF_ERROR(parse_page_header());
+ if (_page_reader->is_header_v2() || !IN_COLLECTION) {
+ if (_page_reader->start_row() <= left_row && left_row <
_page_reader->end_row()) {
+ RETURN_IF_ERROR(load_page_data());
+ // this page contain this row.
+ RETURN_IF_ERROR(_skip_nested_rows_in_page(left_row -
_current_row));
+ _current_row = left_row;
+ break;
+ }
+
+ _current_row = _page_reader->end_row();
+ if (has_next_page()) [[likely]] {
+ RETURN_IF_ERROR(next_page());
+ } else {
+ return Status::InternalError("no match seek row {},
current row {}", left_row,
+ _current_row);
+ }
+ } else {
+ RETURN_IF_ERROR(load_page_data());
+ std::vector<level_t> rep_levels;
+ std::vector<level_t> def_levels;
+ bool cross_page = false;
+
+ size_t result_rows = 0;
+ RETURN_IF_ERROR(load_page_nested_rows(rep_levels, left_row -
_current_row,
Review Comment:
[P1] Reject orphan continuations on the first nested page
On the initial sequential V1 nested read, both logical cursors are zero, so
this seeks with a zero-row frontier. If the first page begins with repetition
levels `[1, 0]`, `load_page_nested_rows()` consumes the leading continuation,
these next lines skip its definition/value payload, and the `0` is rewound as
the first row start. The ordinary read can then satisfy the footer row count
while silently dropping the orphan value. A nonzero level may start a later
page only when a row began on a previous page; that state is impossible for the
ColumnChunk's first data page. Track whether a row has actually started and
return corruption for a leading orphan, with malformed-first-page and
valid-cross-page tests.
--
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]
