github-actions[bot] commented on code in PR #65674:
URL: https://github.com/apache/doris/pull/65674#discussion_r3610485299
##########
be/src/format_v2/parquet/parquet_statistics.cpp:
##########
@@ -762,435 +500,406 @@ void accumulate_zonemap_stats(const ZoneMapEvalContext&
ctx, ParquetPruningStats
} // namespace
+bool can_use_parquet_page_index(const format::FileScanRequest& request,
+ const RuntimeState* runtime_state) {
+ return config::enable_parquet_page_index &&
has_expr_zonemap_filter(request, runtime_state);
+}
+
std::shared_ptr<segment_v2::ZoneMap> ParquetStatisticsUtils::MakeZoneMap(
const ParquetColumnStatistics& statistics) {
return make_zonemap_from_statistics(statistics);
}
ParquetColumnStatistics ParquetStatisticsUtils::TransformColumnStatistics(
- const ParquetColumnSchema& column_schema,
- const std::shared_ptr<::parquet::Statistics>& statistics, const
cctz::time_zone* timezone) {
+ const ParquetColumnSchema& column_schema, const tparquet::Statistics*
statistics,
+ int64_t column_value_count, const cctz::time_zone* timezone) {
ParquetColumnStatistics result;
- if (statistics == nullptr) {
+ if (statistics == nullptr || column_value_count < 0) {
return result;
}
- result.has_null = !statistics->HasNullCount() || statistics->null_count()
> 0;
- result.has_not_null = statistics->num_values() > 0 ||
statistics->HasMinMax();
- result.has_null_count = statistics->HasNullCount();
- if (!result.has_not_null || !statistics->HasMinMax()) {
+ if (statistics->__isset.null_count && statistics->null_count >
column_value_count) {
+ // An impossible null count makes all derived min/max and all-null
flags untrustworthy;
+ // disable pruning instead of turning corrupt footer metadata into
false negatives.
return result;
}
- DORIS_CHECK(column_schema.type != nullptr);
- switch (statistics->physical_type()) {
- case ::parquet::Type::BOOLEAN:
- result.has_min_max = set_decoded_min_max<::parquet::BooleanType>(
- statistics, column_schema, DecodedValueKind::BOOL, &result,
timezone);
- return result;
- case ::parquet::Type::INT32:
- result.has_min_max = set_decoded_min_max<::parquet::Int32Type>(
- statistics, column_schema,
decoded_value_kind(column_schema.type_descriptor),
- &result, timezone);
- return result;
- case ::parquet::Type::INT64:
- result.has_min_max = set_decoded_min_max<::parquet::Int64Type>(
- statistics, column_schema,
decoded_value_kind(column_schema.type_descriptor),
- &result, timezone);
- return result;
- case ::parquet::Type::FLOAT:
- result.has_min_max = set_decoded_min_max<::parquet::FloatType>(
- statistics, column_schema, DecodedValueKind::FLOAT, &result,
timezone);
- return result;
- case ::parquet::Type::DOUBLE:
- result.has_min_max = set_decoded_min_max<::parquet::DoubleType>(
- statistics, column_schema, DecodedValueKind::DOUBLE, &result,
timezone);
- return result;
- case ::parquet::Type::BYTE_ARRAY:
- case ::parquet::Type::FIXED_LEN_BYTE_ARRAY:
- result.has_min_max = set_string_min_max(statistics, column_schema,
&result, timezone);
- return result;
- default:
- return result;
+ const bool has_null_count = statistics->__isset.null_count &&
statistics->null_count >= 0;
+ const int64_t null_count = has_null_count ? statistics->null_count : 0;
+ const bool has_not_null = has_null_count ? column_value_count > null_count
: true;
+ const std::string* min_value = statistics->__isset.min_value
+ ? &statistics->min_value
+ : (statistics->__isset.min ?
&statistics->min : nullptr);
+ const std::string* max_value = statistics->__isset.max_value
+ ? &statistics->max_value
+ : (statistics->__isset.max ?
&statistics->max : nullptr);
+
+ tparquet::ColumnIndex index;
+ index.__set_null_pages({!has_not_null});
+ index.__set_null_counts({null_count});
+ if (min_value != nullptr && max_value != nullptr) {
+ index.__set_min_values({*min_value});
+ index.__set_max_values({*max_value});
+ }
+ // Footer statistics and page indexes share the same little-endian
physical encoding. Reusing
+ // one decoder keeps native row-group and page pruning identical for
logical types and NaNs.
+ if (!build_native_page_statistics(index, column_schema, 0, &result,
timezone)) {
+ return {};
+ }
+ if (!has_null_count) {
+ result.has_null_count = false;
+ result.has_null = true;
}
+ return result;
}
-bool ParquetStatisticsUtils::BloomFilterExcludes(const ParquetColumnSchema&
column_schema,
- int slot_index, const
VExprContextSPtrs& conjuncts,
- const ::parquet::BloomFilter&
bloom_filter) {
- return bloom_filter_excludes(column_schema, slot_index, conjuncts,
bloom_filter);
+bool ParquetStatisticsUtils::NativeBloomFilterExcludes(
+ const ParquetColumnSchema& column_schema, int slot_index,
+ const VExprContextSPtrs& conjuncts, const segment_v2::BloomFilter&
bloom_filter) {
+ if (!bloom_filter_supported(column_schema)) {
+ return false;
+ }
+ NativeParquetBloomFilterAdapter adapter(column_schema, bloom_filter);
+ BloomFilterEvalContext ctx;
+ ctx.slots.emplace(slot_index, BloomFilterEvalContext::SlotBloomFilter {
+ .data_type = column_schema.type,
+ .bloom_filter = &adapter,
+ });
+ return VExprContext::evaluate_bloom_filter(conjuncts, ctx) ==
ZoneMapFilterResult::kNoMatch;
}
namespace {
-ParquetRowGroupPruneReason dictionary_prune_reason(
- const ::parquet::RowGroupMetaData& row_group,
::parquet::ParquetFileReader* file_reader,
- int row_group_idx, const
std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema,
- const format::FileScanRequest& request) {
- const auto conjuncts_by_slot = collect_conjuncts_by_single_slot(
- request.conjuncts, expr_zonemap::single_slot_dictionary_index);
- for (const auto& [slot_index, conjuncts] : conjuncts_by_slot) {
- const auto file_column_id = file_column_id_by_block_position(request,
slot_index);
- if (!file_column_id.has_value()) {
- continue;
- }
- const auto* column_schema = resolve_local_leaf_schema(file_schema,
*file_column_id);
- if (column_schema == nullptr || column_schema->type == nullptr) {
- continue;
- }
- DCHECK_LT(column_schema->leaf_column_id, row_group.num_columns());
- auto column_chunk =
row_group.ColumnChunk(column_schema->leaf_column_id);
- if (column_chunk == nullptr ||
- !supports_dictionary_pruning(*column_schema, *column_chunk) ||
- !is_dictionary_encoded_chunk(*column_chunk)) {
- continue;
- }
-
- ParquetDictionaryWords dict_words;
- if (!read_dictionary_words(file_reader, row_group_idx,
column_schema->leaf_column_id,
- *column_schema, &dict_words)) {
- continue;
- }
- DictionaryEvalContext ctx;
- ctx.slots.emplace(slot_index, DictionaryEvalContext::SlotDictionary {
- .data_type = column_schema->type,
- .values =
dictionary_fields_from_words(dict_words),
- });
- if (VExprContext::evaluate_dictionary_filter(conjuncts, ctx) ==
- ZoneMapFilterResult::kNoMatch) {
- return ParquetRowGroupPruneReason::DICTIONARY;
+void collect_filtered_leaf_ids(const ParquetColumnSchema& column_schema,
+ const format::LocalColumnIndex* projection,
+ std::set<int>* leaf_column_ids) {
+ if (column_schema.kind == ParquetColumnSchemaKind::PRIMITIVE) {
+ if (column_schema.leaf_column_id >= 0) {
+ leaf_column_ids->insert(column_schema.leaf_column_id);
}
+ return;
}
- return ParquetRowGroupPruneReason::NONE;
-}
-
-ParquetRowGroupPruneReason bloom_filter_prune_reason(
- int row_group_idx, const
std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema,
- const format::FileScanRequest& request, RowGroupBloomFilterCache*
bloom_filter_cache,
- ParquetPruningStats* pruning_stats) {
- if (bloom_filter_cache == nullptr) {
- return ParquetRowGroupPruneReason::NONE;
- }
- const auto conjuncts_by_slot = collect_conjuncts_by_single_slot(
- request.conjuncts, expr_zonemap::single_slot_bloom_filter_index);
- for (const auto& [slot_index, conjuncts] : conjuncts_by_slot) {
- const auto file_column_id = file_column_id_by_block_position(request,
slot_index);
- if (!file_column_id.has_value()) {
- continue;
- }
- const auto* column_schema = resolve_local_leaf_schema(file_schema,
*file_column_id);
- if (column_schema == nullptr || column_schema->type == nullptr ||
- !bloom_filter_supported(*column_schema)) {
- continue;
- }
- auto* bloom_filter = bloom_filter_cache->get(row_group_idx,
column_schema->leaf_column_id,
- pruning_stats);
- if (bloom_filter == nullptr) {
+ for (const auto& child_schema : column_schema.children) {
+ if (!format::is_child_projected(projection, child_schema->local_id)) {
continue;
}
- if (ParquetStatisticsUtils::BloomFilterExcludes(*column_schema,
slot_index, conjuncts,
- *bloom_filter)) {
- return ParquetRowGroupPruneReason::BLOOM_FILTER;
- }
+ collect_filtered_leaf_ids(*child_schema,
+ format::find_child_projection(projection,
child_schema->local_id),
+ leaf_column_ids);
}
- return ParquetRowGroupPruneReason::NONE;
}
-void init_bloom_filter_cache(::parquet::ParquetFileReader* file_reader, bool
enable_bloom_filter,
- RowGroupBloomFilterCache* bloom_filter_cache) {
- DORIS_CHECK(bloom_filter_cache != nullptr);
- if (!enable_bloom_filter || file_reader == nullptr) {
- return;
- }
- try {
- bloom_filter_cache->bloom_filter_reader =
&file_reader->GetBloomFilterReader();
- } catch (const ::parquet::ParquetException&) {
- bloom_filter_cache->bloom_filter_reader = nullptr;
- } catch (const std::exception&) {
- bloom_filter_cache->bloom_filter_reader = nullptr;
- }
+bool native_metadata_predicate_is_type_safe(const ParquetColumnSchema&
column_schema) {
+ DORIS_CHECK(column_schema.type != nullptr);
+ // Raw VARBINARY file slots may feed table-side STRING casts. Footer/page
metadata is still in
+ // the pre-cast domain, so using it for a rewritten table predicate can
cause false negatives.
+ return remove_nullable(column_schema.type)->get_primitive_type() !=
TYPE_VARBINARY;
}
-bool check_statistics(const ::parquet::RowGroupMetaData& row_group,
- const std::vector<std::unique_ptr<ParquetColumnSchema>>&
file_schema,
- const format::FileScanRequest& request,
ParquetPruningStats* pruning_stats,
- const cctz::time_zone* timezone) {
+bool check_native_statistics(const tparquet::RowGroup& row_group,
+ const
std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema,
+ const format::FileScanRequest& request,
+ ParquetPruningStats* pruning_stats, const
cctz::time_zone* timezone) {
const auto slot_indexes =
collect_expr_zonemap_slot_indexes(request.conjuncts);
if (slot_indexes.empty()) {
return false;
}
-
ZoneMapEvalContext ctx;
for (const int slot_index : slot_indexes) {
const auto file_column_id = file_column_id_by_block_position(request,
slot_index);
if (!file_column_id.has_value()) {
continue;
}
const auto* column_schema = resolve_local_leaf_schema(file_schema,
*file_column_id);
- if (column_schema == nullptr || column_schema->type == nullptr) {
+ if (column_schema == nullptr || column_schema->type == nullptr ||
+ !native_metadata_predicate_is_type_safe(*column_schema) ||
+ column_schema->leaf_column_id >=
static_cast<int>(row_group.columns.size())) {
continue;
}
-
+ const auto& chunk = row_group.columns[column_schema->leaf_column_id];
std::shared_ptr<segment_v2::ZoneMap> zone_map;
- DCHECK_LT(column_schema->leaf_column_id, row_group.num_columns());
- auto column_chunk =
row_group.ColumnChunk(column_schema->leaf_column_id);
- if (column_chunk != nullptr) {
+ if (chunk.__isset.meta_data) {
+ const auto& column_metadata = chunk.meta_data;
+ const auto* statistics =
+ column_metadata.__isset.statistics ?
&column_metadata.statistics : nullptr;
+ if (statistics != nullptr &&
!detail::can_use_native_footer_min_max(
+
column_schema->type_descriptor, *statistics)) {
+ statistics = nullptr;
+ }
zone_map = ParquetStatisticsUtils::MakeZoneMap(
ParquetStatisticsUtils::TransformColumnStatistics(
- *column_schema, column_chunk->statistics(),
timezone));
+ *column_schema, statistics,
column_metadata.num_values, timezone));
}
add_slot_zonemap(&ctx, slot_index, column_schema->type,
std::move(zone_map));
}
-
const auto result =
VExprContext::evaluate_zonemap_filter(request.conjuncts, ctx);
accumulate_zonemap_stats(ctx, pruning_stats);
return result == ZoneMapFilterResult::kNoMatch;
}
-Status select_row_groups_by_metadata_impl(
- const ::parquet::FileMetaData& metadata, ::parquet::ParquetFileReader*
file_reader,
- const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema,
- const format::FileScanRequest& request, const std::vector<int>*
candidate_row_groups,
- std::vector<int>* selected_row_groups, bool enable_bloom_filter,
- ParquetPruningStats* pruning_stats, const cctz::time_zone* timezone,
- const RuntimeState* runtime_state) {
- int64_t row_group_filter_time_sink = 0;
- SCOPED_RAW_TIMER(pruning_stats == nullptr ? &row_group_filter_time_sink
- :
&pruning_stats->row_group_filter_time);
- if (selected_row_groups == nullptr) {
- return Status::InvalidArgument("selected_row_groups is null");
- }
- selected_row_groups->clear();
+bool is_native_dictionary_data_encoding(tparquet::Encoding::type encoding) {
+ return encoding == tparquet::Encoding::PLAIN_DICTIONARY ||
+ encoding == tparquet::Encoding::RLE_DICTIONARY;
+}
- const int num_row_groups = metadata.num_row_groups();
- const auto candidate_size = candidate_row_groups == nullptr
- ? static_cast<size_t>(num_row_groups)
- : candidate_row_groups->size();
- if (pruning_stats != nullptr) {
- // Scan-range ownership is decided before metadata pruning. Count only
row groups owned by
- // this split so a file divided into multiple splits does not report
the full-file total and
- // out-of-split groups once per split.
- pruning_stats->total_row_groups = cast_set<int64_t>(candidate_size);
+bool is_native_level_encoding(tparquet::Encoding::type encoding) {
+ return encoding == tparquet::Encoding::RLE || encoding ==
tparquet::Encoding::BIT_PACKED;
+}
+
+bool is_native_dictionary_encoded_chunk(const tparquet::ColumnMetaData&
metadata) {
+ if (!metadata.__isset.dictionary_page_offset ||
metadata.dictionary_page_offset < 0) {
+ return false;
}
- selected_row_groups->reserve(candidate_size);
- RowGroupBloomFilterCache bloom_filter_cache;
- init_bloom_filter_cache(file_reader, enable_bloom_filter,
&bloom_filter_cache);
- for (size_t candidate_idx = 0; candidate_idx < candidate_size;
++candidate_idx) {
- const int row_group_idx = candidate_row_groups == nullptr
- ? static_cast<int>(candidate_idx)
- :
(*candidate_row_groups)[candidate_idx];
- DORIS_CHECK(row_group_idx >= 0);
- DORIS_CHECK(row_group_idx < num_row_groups);
- auto row_group = metadata.RowGroup(row_group_idx);
- if (row_group == nullptr) {
- selected_row_groups->push_back(row_group_idx);
- continue;
+ if (metadata.__isset.encoding_stats && !metadata.encoding_stats.empty()) {
+ bool has_dictionary_data_page = false;
+ for (const auto& encoding_stat : metadata.encoding_stats) {
+ if ((encoding_stat.page_type != tparquet::PageType::DATA_PAGE &&
+ encoding_stat.page_type != tparquet::PageType::DATA_PAGE_V2)
||
+ encoding_stat.count <= 0) {
+ continue;
+ }
+ if (!is_native_dictionary_data_encoding(encoding_stat.encoding)) {
+ return false;
+ }
+ has_dictionary_data_page = true;
}
- ParquetRowGroupPruneReason prune_reason =
ParquetRowGroupPruneReason::NONE;
- if (has_expr_zonemap_filter(request, runtime_state) &&
- check_statistics(*row_group, file_schema, request, pruning_stats,
timezone)) {
- prune_reason = ParquetRowGroupPruneReason::STATISTICS;
+ return has_dictionary_data_page;
+ }
+ bool has_dictionary_encoding = false;
+ for (const auto encoding : metadata.encodings) {
+ if (is_native_dictionary_data_encoding(encoding)) {
+ has_dictionary_encoding = true;
+ } else if (!is_native_level_encoding(encoding)) {
+ return false;
}
+ }
+ return has_dictionary_encoding;
+}
- if (prune_reason == ParquetRowGroupPruneReason::NONE) {
- prune_reason = dictionary_prune_reason(*row_group, file_reader,
row_group_idx,
- file_schema, request);
- if (prune_reason == ParquetRowGroupPruneReason::NONE) {
- prune_reason = bloom_filter_prune_reason(row_group_idx,
file_schema, request,
- &bloom_filter_cache,
pruning_stats);
- }
+const format::LocalColumnIndex* find_request_projection(const
format::FileScanRequest& request,
+ format::LocalColumnId
file_column_id) {
+ for (const auto& projection : request.predicate_columns) {
+ if (projection.local_id() == file_column_id.value()) {
+ return &projection;
}
-
- if (prune_reason != ParquetRowGroupPruneReason::NONE) {
- if (pruning_stats != nullptr) {
- pruning_stats->filtered_group_rows += row_group->num_rows();
- if (prune_reason == ParquetRowGroupPruneReason::STATISTICS) {
- ++pruning_stats->filtered_row_groups_by_statistics;
- } else if (prune_reason ==
ParquetRowGroupPruneReason::DICTIONARY) {
- ++pruning_stats->filtered_row_groups_by_dictionary;
- } else if (prune_reason ==
ParquetRowGroupPruneReason::BLOOM_FILTER) {
- ++pruning_stats->filtered_row_groups_by_bloom_filter;
- }
- }
- continue;
+ }
+ for (const auto& projection : request.non_predicate_columns) {
+ if (projection.local_id() == file_column_id.value()) {
+ return &projection;
}
- selected_row_groups->push_back(row_group_idx);
}
- return Status::OK();
+ return nullptr;
}
-} // namespace
-
-Status select_row_groups_by_metadata(
- const ::parquet::FileMetaData& metadata, ::parquet::ParquetFileReader*
file_reader,
+ParquetRowGroupPruneReason native_dictionary_prune_reason(
+ const tparquet::RowGroup& row_group, int row_group_idx,
const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema,
- const format::FileScanRequest& request, const std::vector<int>*
candidate_row_groups,
- std::vector<int>* selected_row_groups, bool enable_bloom_filter,
- ParquetPruningStats* pruning_stats, const cctz::time_zone* timezone,
- const RuntimeState* runtime_state) {
- return select_row_groups_by_metadata_impl(
- metadata, file_reader, file_schema, request, candidate_row_groups,
selected_row_groups,
- enable_bloom_filter, pruning_stats, timezone, runtime_state);
-}
-
-namespace {
-
-template <typename ParquetDType>
-bool set_page_decoded_min_max(const std::shared_ptr<::parquet::ColumnIndex>&
column_index,
- const ParquetColumnSchema& column_schema, size_t
page_idx,
- DecodedValueKind value_kind,
ParquetColumnStatistics* page_statistics,
- const cctz::time_zone* timezone) {
- const auto typed_index =
-
std::static_pointer_cast<::parquet::TypedColumnIndex<ParquetDType>>(column_index);
- if (page_idx >= typed_index->min_values().size() ||
- page_idx >= typed_index->max_values().size()) {
- return false;
+ const format::FileScanRequest& request, const cctz::time_zone*
timezone,
+ ParquetFileContext* file_context) {
+ if (file_context == nullptr || file_context->native_metadata == nullptr) {
+ return ParquetRowGroupPruneReason::NONE;
}
- const auto& min_value = typed_index->min_values()[page_idx];
- const auto& max_value = typed_index->max_values()[page_idx];
- if constexpr (std::is_same_v<ParquetDType, ::parquet::Int64Type>) {
- if (!timestamp_min_max_is_safe(column_schema, min_value, max_value,
timezone)) {
- return false;
+ const auto conjuncts_by_slot = collect_conjuncts_by_single_slot(
+ request.conjuncts, expr_zonemap::single_slot_dictionary_index);
+ for (const auto& [slot_index, conjuncts] : conjuncts_by_slot) {
+ const auto file_column_id = file_column_id_by_block_position(request,
slot_index);
+ if (!file_column_id.has_value()) {
+ continue;
+ }
+ const auto* column_schema = resolve_local_leaf_schema(file_schema,
*file_column_id);
+ const auto* projection = find_request_projection(request,
*file_column_id);
+ if (column_schema == nullptr || projection == nullptr ||
column_schema->type == nullptr ||
+ !column_schema->type_descriptor.is_string_like ||
+ column_schema->leaf_column_id >=
static_cast<int>(row_group.columns.size())) {
+ continue;
+ }
+ if (!native_metadata_predicate_is_type_safe(*column_schema)) {
+ // The file-local VARBINARY may feed a table-side STRING cast.
Pruning before that cast
+ // can compare different Field kinds and incorrectly discard a
matching row group.
+ continue;
+ }
+ const auto& chunk = row_group.columns[column_schema->leaf_column_id];
+ if (!chunk.__isset.meta_data ||
+ (chunk.meta_data.type != tparquet::Type::BYTE_ARRAY &&
+ chunk.meta_data.type != tparquet::Type::FIXED_LEN_BYTE_ARRAY) ||
+ !is_native_dictionary_encoded_chunk(chunk.meta_data)) {
+ continue;
+ }
+ std::unique_ptr<ParquetColumnReader> reader;
+ const std::vector<RowRange> ranges {{0, row_group.num_rows}};
+ const std::unordered_map<int, tparquet::OffsetIndex> offset_indexes;
+ const auto status = NativeColumnReader::create(
+ *column_schema, projection, file_context->native_file,
+ file_context->native_metadata, row_group_idx, ranges,
offset_indexes, timezone,
+ file_context->native_io_ctx, nullptr,
file_context->native_page_cache_enabled,
+ file_context->native_page_cache_file_key, true, {}, &reader);
Review Comment:
[P2] Publish the dictionary-pruning probe's native statistics. This reader
receives an empty `ParquetColumnReaderProfile`, and its construction
initializes the chunk reader and reads/caches/decompresses/decodes the
dictionary page before `dictionary_values()` materializes that dictionary; its
destructor can sync only to null counters. The removed file adapter accumulated
page-cache activity at file scope, but `ParquetFileContext::page_cache_stats()`
now returns `{}`, so a row group eliminated here can perform cold or warm page
work while `PageReadCount`, `PageCache*`, header/decompression, and
dictionary-decode counters remain zero. Pass the column-reader profile into
planning (or explicitly aggregate the probe before destruction), with cold/warm
dictionary-pruned profile coverage.
##########
be/src/core/data_type_serde/parquet_decode_source.h:
##########
@@ -0,0 +1,273 @@
+// 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.
+
+#pragma once
+
+#include <cstddef>
+#include <cstdint>
+#include <limits>
+#include <vector>
+
+#include "common/check.h"
+#include "common/status.h"
+#include "core/column/column.h"
+#include "core/string_ref.h"
+
+namespace cctz {
+class time_zone;
+} // namespace cctz
+
+namespace doris {
+
+// These enums deliberately do not expose parquet thrift classes to the core
type system. The
+// format reader translates the thrift metadata once when it creates a column
reader.
+enum class ParquetPhysicalType {
+ BOOLEAN,
+ INT32,
+ INT64,
+ INT96,
+ FLOAT,
+ DOUBLE,
+ BYTE_ARRAY,
+ FIXED_LEN_BYTE_ARRAY,
+};
+
+enum class ParquetValueEncoding {
+ PLAIN,
+ DICTIONARY,
+ RLE,
+ BIT_PACKED,
+ DELTA_BINARY_PACKED,
+ DELTA_LENGTH_BYTE_ARRAY,
+ DELTA_BYTE_ARRAY,
+ BYTE_STREAM_SPLIT,
+};
+
+enum class ParquetTimeUnit {
+ UNKNOWN,
+ MILLIS,
+ MICROS,
+ NANOS,
+};
+
+enum class ParquetLogicalType {
+ NONE,
+ STRING,
+ DECIMAL,
+ DATE,
+ TIME,
+ TIMESTAMP,
+ INTEGER,
+ UUID,
+ FLOAT16,
+};
+
+// Immutable metadata required to turn one Parquet physical value into the
selected Doris type.
+// Encoding describes how the value source is read; logical annotations
describe its meaning.
+struct ParquetDecodeContext {
+ ParquetPhysicalType physical_type = ParquetPhysicalType::INT32;
+ ParquetValueEncoding encoding = ParquetValueEncoding::PLAIN;
+ ParquetLogicalType logical_type = ParquetLogicalType::NONE;
+ ParquetTimeUnit time_unit = ParquetTimeUnit::UNKNOWN;
+
+ int32_t type_length = -1;
+ int32_t decimal_precision = -1;
+ int32_t decimal_scale = -1;
+ int32_t logical_integer_bit_width = -1;
+ bool logical_integer_is_signed = true;
+ bool timestamp_is_adjusted_to_utc = false;
+ bool logical_float16 = false;
+ bool logical_uuid = false;
+ bool dictionary_index_only = false;
+
+ const cctz::time_zone* timezone = nullptr;
+};
+
+struct ParquetSelectionRange {
+ size_t first = 0;
+ size_t count = 0;
+};
+
+// A decoder may produce multiple contiguous spans for one request (for
example delta encodings).
+// Consumers are invoked per span, never per value, keeping virtual dispatch
out of the row loop.
+class ParquetFixedValueConsumer {
+public:
+ virtual ~ParquetFixedValueConsumer() = default;
+ virtual Status consume(const uint8_t* values, size_t num_values, size_t
value_width) = 0;
+ virtual Status consume_selected(const uint8_t* values, size_t value_width,
+ const std::vector<ParquetSelectionRange>&
ranges) {
+ for (const auto& range : ranges) {
+ RETURN_IF_ERROR(consume(values + range.first * value_width,
range.count, value_width));
+ }
+ return Status::OK();
+ }
+};
+
+class ParquetBinaryValueConsumer {
+public:
+ virtual ~ParquetBinaryValueConsumer() = default;
+ virtual Status consume(const StringRef* values, size_t num_values) = 0;
+
+ // PLAIN BYTE_ARRAY decoders already have to parse every length prefix.
Publish the parsed
+ // source and destination offsets so string columns do not rebuild an
equally large StringRef
+ // array and rescan all lengths before copying. Spans are expressed in
output coordinates and
+ // preserve adjacent surviving runs for consumers that can amortize range
setup.
+ virtual Status consume_plain_byte_array(const char* encoded_data,
+ const uint32_t* payload_offsets,
+ const uint32_t* value_offsets,
size_t num_values,
+ const
std::vector<ParquetSelectionRange>& value_spans) {
+ std::vector<StringRef> values;
+ values.reserve(num_values);
+ for (size_t row = 0; row < num_values; ++row) {
+ values.emplace_back(encoded_data + payload_offsets[row],
+ value_offsets[row + 1] - value_offsets[row]);
+ }
+ return consume(values.data(), values.size());
+ }
+};
+
+// Physical value ranges selected from one page-bounded decode request.
Definition-level NULLs are
+// intentionally excluded: the native ColumnReader uses this plan only when
the batch has no NULL
+// leaf slots, so selected values can be appended in one pass without a
temporary nullable column.
+// Ranges are sorted, disjoint, and expressed in the physical value stream's
coordinate space.
+struct ParquetSelection {
+ size_t total_values = 0;
+ size_t selected_values = 0;
+ std::vector<ParquetSelectionRange> ranges;
+};
+
+// Encoding decoders implement this interface. They own encoded-stream cursors
and dictionary
+// storage, but they never know the destination Doris column type.
DataTypeSerDe owns the consumer
+// and therefore the physical/logical-to-Doris conversion.
+class ParquetDecodeSource {
+public:
+ virtual ~ParquetDecodeSource() = default;
+
+ virtual Status decode_fixed_values(size_t num_values,
ParquetFixedValueConsumer& consumer) = 0;
+ virtual Status decode_binary_values(size_t num_values,
+ ParquetBinaryValueConsumer& consumer)
= 0;
+ virtual Status skip_values(size_t num_values) = 0;
+
+ // Batch-level sparse decode. The default implementation preserves every
encoding's cursor
+ // semantics while moving SerDe dispatch and consumer construction out of
the selection-run
+ // loop. Decoders with cheap random access or batch decode override these
methods to remove the
+ // remaining per-range virtual calls as well.
+ virtual Status decode_selected_fixed_values(const ParquetSelection&
selection,
+ ParquetFixedValueConsumer&
consumer) {
+ size_t cursor = 0;
+ for (const auto& range : selection.ranges) {
+ DORIS_CHECK(range.first >= cursor);
+ DORIS_CHECK(range.first + range.count <= selection.total_values);
+ RETURN_IF_ERROR(skip_values(range.first - cursor));
+ RETURN_IF_ERROR(decode_fixed_values(range.count, consumer));
+ cursor = range.first + range.count;
+ }
+ return skip_values(selection.total_values - cursor);
+ }
+ virtual Status decode_selected_binary_values(const ParquetSelection&
selection,
+ ParquetBinaryValueConsumer&
consumer) {
+ size_t cursor = 0;
+ for (const auto& range : selection.ranges) {
+ DORIS_CHECK(range.first >= cursor);
+ DORIS_CHECK(range.first + range.count <= selection.total_values);
+ RETURN_IF_ERROR(skip_values(range.first - cursor));
+ RETURN_IF_ERROR(decode_binary_values(range.count, consumer));
+ cursor = range.first + range.count;
+ }
+ return skip_values(selection.total_values - cursor);
+ }
+
+ virtual bool has_dictionary() const { return false; }
+ virtual uint64_t dictionary_generation() const { return 0; }
+ virtual size_t dictionary_size() const { return 0; }
+ virtual Status decode_dictionary(ParquetFixedValueConsumer& fixed_consumer,
+ ParquetBinaryValueConsumer&
binary_consumer) {
+ return Status::NotSupported("Parquet dictionary is not supported by
this decoder");
+ }
+ virtual Status decode_dictionary_indices(size_t num_values,
std::vector<uint32_t>* indices) {
+ return Status::NotSupported("Parquet dictionary indices are not
supported by this decoder");
+ }
+ virtual Status decode_selected_dictionary_indices(const ParquetSelection&
selection,
+ std::vector<uint32_t>*
indices) {
+ DORIS_CHECK(indices != nullptr);
+ indices->clear();
+ indices->reserve(selection.selected_values);
+ std::vector<uint32_t> range_indices;
+ size_t cursor = 0;
+ for (const auto& range : selection.ranges) {
+ DORIS_CHECK(range.first >= cursor);
+ DORIS_CHECK(range.first + range.count <= selection.total_values);
+ RETURN_IF_ERROR(skip_values(range.first - cursor));
+ RETURN_IF_ERROR(decode_dictionary_indices(range.count,
&range_indices));
+ indices->insert(indices->end(), range_indices.begin(),
range_indices.end());
+ cursor = range.first + range.count;
+ }
+ RETURN_IF_ERROR(skip_values(selection.total_values - cursor));
+ DORIS_CHECK_EQ(indices->size(), selection.selected_values);
+ return Status::OK();
+ }
+};
+
+// Dictionary values are materialized once into the selected Doris type. The
state belongs to a
+// column reader rather than DataTypeSerDe because a SerDe instance can be
shared by many files.
+struct ParquetMaterializationState {
+ MutableColumnPtr typed_dictionary;
+ std::vector<uint32_t> dictionary_indices;
+ ParquetSelection selection;
+ uint64_t dictionary_generation = std::numeric_limits<uint64_t>::max();
+ bool enable_strict_mode = false;
+ IColumn::Filter* conversion_failure_null_map = nullptr;
+ IColumn::Filter dictionary_conversion_failures;
+
+ void reset_dictionary() {
+ typed_dictionary.reset();
+ dictionary_indices.clear();
+ dictionary_conversion_failures.clear();
+ dictionary_generation = std::numeric_limits<uint64_t>::max();
+ }
+
+ bool can_insert_null_on_conversion_failure() const {
+ return !enable_strict_mode && conversion_failure_null_map != nullptr;
+ }
+
+ bool mark_conversion_failure(size_t output_row) {
+ if (!can_insert_null_on_conversion_failure()) {
+ return false;
+ }
+ DORIS_CHECK_LT(output_row, conversion_failure_null_map->size());
+ (*conversion_failure_null_map)[output_row] = 1;
+ return true;
+ }
+
+ IColumn::Filter* begin_dictionary_conversion(size_t dictionary_size) {
Review Comment:
[P1] Defer dictionary conversion errors until an ID references them. This
redirects per-entry failures only for nullable non-strict output, but every
direct numeric/date/time/timestamp/decimal SerDe converts the complete
dictionary before decoding page IDs. Thus a valid physical INT32 dictionary
`[1, 1000]` with data IDs `[0]`, read as strict or non-nullable TINYINT, now
fails on the unused `1000` even though the only materialized row is `1`; the
removed path converted expanded, referenced values. Keep a
cardinality-preserving placeholder plus per-entry conversion status in every
mode, and raise the strict/non-nullable error only when a decoded/selected ID
references that entry. Please cover both unused-bad and referenced-bad entries.
--
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]
