This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new e4f23b9ac56 branch-4.2: [improvement](be) Reuse Parquet profile
counters across external scan files (#69300) (#69804)
e4f23b9ac56 is described below
commit e4f23b9ac56b5c114bd2e7f02777c80ee90c9d68
Author: 924060929 <[email protected]>
AuthorDate: Thu Oct 29 18:31:44 2026 +0800
branch-4.2: [improvement](be) Reuse Parquet profile counters across
external scan files (#69300) (#69804)
cherry pick from #69300
---
be/src/exec/operator/file_scan_operator.cpp | 10 ++
be/src/exec/operator/file_scan_operator.h | 9 +-
be/src/exec/scan/file_scanner_v2.cpp | 4 +-
be/src/exec/scan/file_scanner_v2.h | 1 +
be/src/format_v2/file_reader.cpp | 2 +-
be/src/format_v2/parquet/parquet_profile.cpp | 75 ++++-----
be/src/format_v2/parquet/parquet_profile.h | 75 ++++++---
be/src/format_v2/parquet/parquet_reader.cpp | 10 +-
be/src/format_v2/parquet/parquet_reader.h | 3 +-
.../parquet/reader/variant_column_reader.cpp | 62 +++++---
be/src/format_v2/table/hudi_reader.cpp | 1 +
.../iceberg_position_delete_sys_table_reader.cpp | 11 ++
.../iceberg_position_delete_sys_table_reader.h | 2 +
be/src/format_v2/table/iceberg_reader.cpp | 2 +-
be/src/format_v2/table/load_reader.cpp | 1 +
be/src/format_v2/table/paimon_reader.cpp | 1 +
be/src/format_v2/table_reader.cpp | 4 +-
be/src/format_v2/table_reader.h | 6 +
be/test/format_v2/parquet/parquet_profile_test.cpp | 169 +++++++++++++++++++++
be/test/format_v2/parquet/parquet_reader_test.cpp | 50 ++++++
.../parquet/variant_column_reader_test.cpp | 8 +-
be/test/format_v2/table/hudi_reader_test.cpp | 74 +++++++++
...eberg_position_delete_sys_table_reader_test.cpp | 45 ++++++
be/test/format_v2/table/iceberg_reader_test.cpp | 37 +++++
be/test/format_v2/table/load_reader_test.cpp | 17 ++-
25 files changed, 577 insertions(+), 102 deletions(-)
diff --git a/be/src/exec/operator/file_scan_operator.cpp
b/be/src/exec/operator/file_scan_operator.cpp
index 9ba2747624b..6802dae071f 100644
--- a/be/src/exec/operator/file_scan_operator.cpp
+++ b/be/src/exec/operator/file_scan_operator.cpp
@@ -34,6 +34,7 @@
#include "exec/scan/scanner_context.h"
#include "format/format_common.h"
#include "format/table/iceberg_scan_semantics.h"
+#include "format_v2/parquet/parquet_profile.h"
#include "storage/storage_engine.h"
#include "storage/tablet/tablet_manager.h"
@@ -63,6 +64,11 @@ bool contains_variant_type(const DataTypePtr& input) {
} // namespace
+FileScanLocalState::FileScanLocalState(RuntimeState* state, OperatorXBase*
parent)
+ : ScanLocalState<FileScanLocalState>(state, parent) {}
+
+FileScanLocalState::~FileScanLocalState() = default;
+
PushDownType FileScanLocalState::_should_push_down_binary_predicate(
VectorizedFnCall* fn_call, VExprContext* expr_ctx, Field& constant_val,
const std::set<std::string> fn_name) const {
@@ -273,6 +279,10 @@ Status
FileScanLocalState::_init_scanners(std::list<ScannerSPtr>* scanners) {
"External VARIANT columns require FileScannerV2; the legacy
file scanner does "
"not support VARIANT");
}
+ if (use_file_scanner_v2) {
+ _parquet_profile_cache =
+
std::make_unique<format::parquet::ParquetProfileCache>(_scanner_profile.get());
+ }
for (int i = 0; i < _max_scanners; ++i) {
ScannerSPtr scanner;
if (use_file_scanner_v2) {
diff --git a/be/src/exec/operator/file_scan_operator.h
b/be/src/exec/operator/file_scan_operator.h
index 80c16e1177d..03c0c980d93 100644
--- a/be/src/exec/operator/file_scan_operator.h
+++ b/be/src/exec/operator/file_scan_operator.h
@@ -28,6 +28,10 @@
#include "exec/scan/split_source_connector.h"
#include "format/format_common.h"
+namespace doris::format::parquet {
+class ParquetProfileCache;
+}
+
namespace doris {
#include "common/compile_check_begin.h"
class FileScanner;
@@ -42,8 +46,8 @@ public:
using Parent = FileScanOperatorX;
using Base = ScanLocalState<FileScanLocalState>;
ENABLE_FACTORY_CREATOR(FileScanLocalState);
- FileScanLocalState(RuntimeState* state, OperatorXBase* parent)
- : ScanLocalState<FileScanLocalState>(state, parent) {}
+ FileScanLocalState(RuntimeState* state, OperatorXBase* parent);
+ ~FileScanLocalState() override;
Status init(RuntimeState* state, LocalStateInfo& info) override;
@@ -101,6 +105,7 @@ private:
VectorizedFnCall* fn_call, VExprContext* expr_ctx, Field&
constant_val,
const std::set<std::string> fn_name) const override;
std::shared_ptr<SplitSourceConnector> _split_source = nullptr;
+ std::unique_ptr<format::parquet::ParquetProfileCache>
_parquet_profile_cache;
int _max_scanners;
// A in memory cache to save some common components
// of the this scan node. eg:
diff --git a/be/src/exec/scan/file_scanner_v2.cpp
b/be/src/exec/scan/file_scanner_v2.cpp
index 45b40ee8ab4..2090a330624 100644
--- a/be/src/exec/scan/file_scanner_v2.cpp
+++ b/be/src/exec/scan/file_scanner_v2.cpp
@@ -415,7 +415,8 @@ FileScannerV2::FileScannerV2(RuntimeState* state,
FileScanLocalState* local_stat
: Scanner(state, local_state, limit, profile),
_split_source(std::move(split_source)),
_kv_cache(kv_cache),
- _constructed_scanners(local_state->_max_scanners) {
+ _constructed_scanners(local_state->_max_scanners),
+ _parquet_profile_cache(local_state->_parquet_profile_cache.get()) {
(void)colname_to_slot_id;
if (state->get_query_ctx() != nullptr &&
state->get_query_ctx()->file_scan_range_params_map.count(local_state->parent_id())
> 0) {
@@ -777,6 +778,7 @@ Status FileScannerV2::_init_table_reader(const
TFileRangeDesc& range) {
.io_ctx = _io_ctx,
.runtime_state = _state,
.scanner_profile = _local_state != nullptr ?
_local_state->scanner_profile() : _profile,
+ .parquet_profile_cache = _parquet_profile_cache,
.file_slot_descs = &_file_slot_descs,
.push_down_agg_type = _local_state != nullptr ?
_local_state->get_push_down_agg_type()
:
TPushAggOp::type::NONE,
diff --git a/be/src/exec/scan/file_scanner_v2.h
b/be/src/exec/scan/file_scanner_v2.h
index b7de342988a..660ae0f4ff0 100644
--- a/be/src/exec/scan/file_scanner_v2.h
+++ b/be/src/exec/scan/file_scanner_v2.h
@@ -261,6 +261,7 @@ private:
std::shared_ptr<io::IOContext> _io_ctx;
ShardedKVCache* _kv_cache = nullptr;
int _constructed_scanners = 1;
+ format::parquet::ParquetProfileCache* _parquet_profile_cache = nullptr;
RuntimeProfile::Counter* _scanner_total_timer = nullptr;
RuntimeProfile::Counter* _init_timer = nullptr;
diff --git a/be/src/format_v2/file_reader.cpp b/be/src/format_v2/file_reader.cpp
index 894cd49fa6b..7ce26e61308 100644
--- a/be/src/format_v2/file_reader.cpp
+++ b/be/src/format_v2/file_reader.cpp
@@ -100,7 +100,7 @@ std::string FileScanRequest::debug_string() const {
}
Status FileReader::init(RuntimeState* state) {
- _init_profile();
+ // Concrete readers initialize their counters before starting their
lifecycle timers.
SCOPED_RAW_TIMER(&_reader_statistics.file_reader_create_time);
++_reader_statistics.open_file_num;
io::FileReaderOptions reader_options =
diff --git a/be/src/format_v2/parquet/parquet_profile.cpp
b/be/src/format_v2/parquet/parquet_profile.cpp
index a9788db044b..4cfb2dad3af 100644
--- a/be/src/format_v2/parquet/parquet_profile.cpp
+++ b/be/src/format_v2/parquet/parquet_profile.cpp
@@ -22,6 +22,20 @@
namespace doris::format::parquet {
+const ParquetProfile& ParquetProfileCache::get() {
+ std::call_once(_initialized, [this] { _counters.init(_profile); });
+ return _counters;
+}
+
+ParquetProfileCache* ParquetProfileCache::get_child_cache(RuntimeProfile*
child_profile) {
+ std::lock_guard lock(_children_mutex);
+ auto& cache = _children[child_profile];
+ if (cache == nullptr) {
+ cache = std::make_unique<ParquetProfileCache>(child_profile);
+ }
+ return cache.get();
+}
+
namespace {
std::shared_ptr<RuntimeProfile::Counter>
add_persistent_counter(RuntimeProfile* profile,
@@ -109,38 +123,40 @@ void ParquetProfile::init(RuntimeProfile* profile) {
ADD_CHILD_TIMER_WITH_LEVEL(profile, "LevelOnlySkipTime",
parquet_profile, 1);
materialization_time =
ADD_CHILD_TIMER_WITH_LEVEL(profile, "MaterializationTime",
parquet_profile, 1);
- variant_reconstruction_time = add_persistent_counter(profile,
"VariantReconstructionTime",
- TUnit::TIME_NS,
parquet_profile);
- variant_reconstructed_rows = add_persistent_counter(profile,
"VariantReconstructedRows",
- TUnit::UNIT,
parquet_profile);
- variant_unshredded_direct_import_time = add_persistent_counter(
+ auto variant = std::make_shared<ParquetVariantProfile>();
+ variant->variant_reconstruction_time = add_persistent_counter(
+ profile, "VariantReconstructionTime", TUnit::TIME_NS,
parquet_profile);
+ variant->variant_reconstructed_rows = add_persistent_counter(
+ profile, "VariantReconstructedRows", TUnit::UNIT, parquet_profile);
+ variant->variant_unshredded_direct_import_time = add_persistent_counter(
profile, "VariantUnshreddedDirectImportTime", TUnit::TIME_NS,
parquet_profile);
- variant_unshredded_direct_import_rows = add_persistent_counter(
+ variant->variant_unshredded_direct_import_rows = add_persistent_counter(
profile, "VariantUnshreddedDirectImportRows", TUnit::UNIT,
parquet_profile);
- variant_unshredded_direct_import_bytes = add_persistent_counter(
+ variant->variant_unshredded_direct_import_bytes = add_persistent_counter(
profile, "VariantUnshreddedDirectImportBytes", TUnit::BYTES,
parquet_profile);
- variant_unshredded_direct_seek_time = add_persistent_counter(
+ variant->variant_unshredded_direct_seek_time = add_persistent_counter(
profile, "VariantUnshreddedDirectSeekTime", TUnit::TIME_NS,
parquet_profile);
- variant_unshredded_direct_seek_rows = add_persistent_counter(
+ variant->variant_unshredded_direct_seek_rows = add_persistent_counter(
profile, "VariantUnshreddedDirectSeekRows", TUnit::UNIT,
parquet_profile);
- variant_unshredded_direct_seek_bytes = add_persistent_counter(
+ variant->variant_unshredded_direct_seek_bytes = add_persistent_counter(
profile, "VariantUnshreddedDirectSeekBytes", TUnit::BYTES,
parquet_profile);
- variant_unshredded_prefix_reuse_rows = add_persistent_counter(
+ variant->variant_unshredded_prefix_reuse_rows = add_persistent_counter(
profile, "VariantUnshreddedPrefixReuseRows", TUnit::UNIT,
parquet_profile);
- variant_direct_subtree_rows = add_persistent_counter(profile,
"VariantDirectSubtreeRows",
- TUnit::UNIT,
parquet_profile);
- variant_direct_leaf_rows =
+ variant->variant_direct_subtree_rows = add_persistent_counter(
+ profile, "VariantDirectSubtreeRows", TUnit::UNIT, parquet_profile);
+ variant->variant_direct_leaf_rows =
add_persistent_counter(profile, "VariantDirectLeafRows",
TUnit::UNIT, parquet_profile);
- variant_direct_leaf_path_misses = add_persistent_counter(profile,
"VariantDirectLeafPathMisses",
- TUnit::UNIT,
parquet_profile);
- variant_direct_leaf_residual_fallbacks = add_persistent_counter(
+ variant->variant_direct_leaf_path_misses = add_persistent_counter(
+ profile, "VariantDirectLeafPathMisses", TUnit::UNIT,
parquet_profile);
+ variant->variant_direct_leaf_residual_fallbacks = add_persistent_counter(
profile, "VariantDirectLeafResidualFallbacks", TUnit::UNIT,
parquet_profile);
- variant_direct_leaf_residual_merged_rows = add_persistent_counter(
+ variant->variant_direct_leaf_residual_merged_rows = add_persistent_counter(
profile, "VariantDirectLeafResidualMergedRows", TUnit::UNIT,
parquet_profile);
- variant_residual_seek_rows = add_persistent_counter(profile,
"VariantResidualSeekRows",
- TUnit::UNIT,
parquet_profile);
- variant_direct_leaf_unsupported_fallbacks = add_persistent_counter(
+ variant->variant_residual_seek_rows = add_persistent_counter(profile,
"VariantResidualSeekRows",
+ TUnit::UNIT,
parquet_profile);
+ variant->variant_direct_leaf_unsupported_fallbacks =
add_persistent_counter(
profile, "VariantDirectLeafUnsupportedFallbacks", TUnit::UNIT,
parquet_profile);
+ variant_counters = std::move(variant);
hybrid_selection_batches = ADD_CHILD_COUNTER_WITH_LEVEL(profile,
"HybridSelectionBatches",
TUnit::UNIT,
parquet_profile, 1);
hybrid_selection_ranges = ADD_CHILD_COUNTER_WITH_LEVEL(profile,
"HybridSelectionRanges",
@@ -369,22 +385,7 @@ ParquetColumnReaderProfile
ParquetProfile::column_reader_profile() const {
.level_only_read_time = level_only_read_time,
.level_only_skip_time = level_only_skip_time,
.materialization_time = materialization_time,
- .variant_reconstruction_time = variant_reconstruction_time,
- .variant_reconstructed_rows = variant_reconstructed_rows,
- .variant_unshredded_direct_import_time =
variant_unshredded_direct_import_time,
- .variant_unshredded_direct_import_rows =
variant_unshredded_direct_import_rows,
- .variant_unshredded_direct_import_bytes =
variant_unshredded_direct_import_bytes,
- .variant_unshredded_direct_seek_time =
variant_unshredded_direct_seek_time,
- .variant_unshredded_direct_seek_rows =
variant_unshredded_direct_seek_rows,
- .variant_unshredded_direct_seek_bytes =
variant_unshredded_direct_seek_bytes,
- .variant_unshredded_prefix_reuse_rows =
variant_unshredded_prefix_reuse_rows,
- .variant_direct_subtree_rows = variant_direct_subtree_rows,
- .variant_direct_leaf_rows = variant_direct_leaf_rows,
- .variant_direct_leaf_path_misses = variant_direct_leaf_path_misses,
- .variant_direct_leaf_residual_fallbacks =
variant_direct_leaf_residual_fallbacks,
- .variant_direct_leaf_residual_merged_rows =
variant_direct_leaf_residual_merged_rows,
- .variant_residual_seek_rows = variant_residual_seek_rows,
- .variant_direct_leaf_unsupported_fallbacks =
variant_direct_leaf_unsupported_fallbacks,
+ .variant_counters = variant_counters,
.hybrid_selection_batches = hybrid_selection_batches,
.hybrid_selection_ranges = hybrid_selection_ranges,
.hybrid_selection_null_fallback_batches =
hybrid_selection_null_fallback_batches,
diff --git a/be/src/format_v2/parquet/parquet_profile.h
b/be/src/format_v2/parquet/parquet_profile.h
index 0a27dc9d8c5..11673ed2de7 100644
--- a/be/src/format_v2/parquet/parquet_profile.h
+++ b/be/src/format_v2/parquet/parquet_profile.h
@@ -16,6 +16,8 @@
#pragma once
#include <memory>
+#include <mutex>
+#include <unordered_map>
#include "runtime/runtime_profile.h"
@@ -32,14 +34,9 @@ struct ParquetPageSkipProfile {
// ============================================================================
// ============================================================================
-struct ParquetColumnReaderProfile {
- RuntimeProfile::Counter* reader_read_rows = nullptr; // rows read by
read()
- RuntimeProfile::Counter* reader_skip_rows = nullptr; // rows skipped by
skip()
- RuntimeProfile::Counter* reader_select_rows = nullptr; // rows selected by
select()
- // COUNT(nullable_col) shape-only path; ordinary scans keep both counters
zero.
- RuntimeProfile::Counter* level_only_read_time = nullptr;
- RuntimeProfile::Counter* level_only_skip_time = nullptr;
- RuntimeProfile::Counter* materialization_time = nullptr; // value
materialization time (ns)
+// Deferred Variant columns can outlive the scanner. This bundle retains all
their counters,
+// while ordinary file/column reader copies only increment one shared
reference count.
+struct ParquetVariantProfile {
std::shared_ptr<RuntimeProfile::Counter> variant_reconstruction_time;
std::shared_ptr<RuntimeProfile::Counter> variant_reconstructed_rows;
// Complete unshredded roots imported without rebuilding their Variant
value tree. Bytes count
@@ -59,6 +56,24 @@ struct ParquetColumnReaderProfile {
std::shared_ptr<RuntimeProfile::Counter>
variant_direct_leaf_residual_merged_rows;
std::shared_ptr<RuntimeProfile::Counter> variant_residual_seek_rows;
std::shared_ptr<RuntimeProfile::Counter>
variant_direct_leaf_unsupported_fallbacks;
+};
+
+struct ParquetColumnReaderProfile {
+ RuntimeProfile::Counter* reader_read_rows = nullptr; // rows read by
read()
+ RuntimeProfile::Counter* reader_skip_rows = nullptr; // rows skipped by
skip()
+ RuntimeProfile::Counter* reader_select_rows = nullptr; // rows selected by
select()
+ // COUNT(nullable_col) shape-only path; ordinary scans keep both counters
zero.
+ RuntimeProfile::Counter* level_only_read_time = nullptr;
+ RuntimeProfile::Counter* level_only_skip_time = nullptr;
+ RuntimeProfile::Counter* materialization_time = nullptr; // value
materialization time (ns)
+ // One ownership reference per profile copy, rather than one per Variant
counter.
+ std::shared_ptr<const ParquetVariantProfile> variant_counters;
+
+ const ParquetVariantProfile& variant_profile() const {
+ static const ParquetVariantProfile disabled;
+ return variant_counters ? *variant_counters : disabled;
+ }
+
RuntimeProfile::Counter* hybrid_selection_batches = nullptr;
RuntimeProfile::Counter* hybrid_selection_ranges = nullptr;
RuntimeProfile::Counter* hybrid_selection_null_fallback_batches = nullptr;
@@ -193,23 +208,14 @@ struct ParquetProfile {
RuntimeProfile::Counter* level_only_read_time = nullptr;
RuntimeProfile::Counter* level_only_skip_time = nullptr;
RuntimeProfile::Counter* materialization_time = nullptr;
- std::shared_ptr<RuntimeProfile::Counter> variant_reconstruction_time;
- std::shared_ptr<RuntimeProfile::Counter> variant_reconstructed_rows;
- // These counters are persistent because direct import can happen after
scanner teardown.
- std::shared_ptr<RuntimeProfile::Counter>
variant_unshredded_direct_import_time;
- std::shared_ptr<RuntimeProfile::Counter>
variant_unshredded_direct_import_rows;
- std::shared_ptr<RuntimeProfile::Counter>
variant_unshredded_direct_import_bytes;
- std::shared_ptr<RuntimeProfile::Counter>
variant_unshredded_direct_seek_time;
- std::shared_ptr<RuntimeProfile::Counter>
variant_unshredded_direct_seek_rows;
- std::shared_ptr<RuntimeProfile::Counter>
variant_unshredded_direct_seek_bytes;
- std::shared_ptr<RuntimeProfile::Counter>
variant_unshredded_prefix_reuse_rows;
- std::shared_ptr<RuntimeProfile::Counter> variant_direct_subtree_rows;
- std::shared_ptr<RuntimeProfile::Counter> variant_direct_leaf_rows;
- std::shared_ptr<RuntimeProfile::Counter> variant_direct_leaf_path_misses;
- std::shared_ptr<RuntimeProfile::Counter>
variant_direct_leaf_residual_fallbacks;
- std::shared_ptr<RuntimeProfile::Counter>
variant_direct_leaf_residual_merged_rows;
- std::shared_ptr<RuntimeProfile::Counter> variant_residual_seek_rows;
- std::shared_ptr<RuntimeProfile::Counter>
variant_direct_leaf_unsupported_fallbacks;
+ // One ownership reference per profile copy, rather than one per Variant
counter.
+ std::shared_ptr<const ParquetVariantProfile> variant_counters;
+
+ const ParquetVariantProfile& variant_profile() const {
+ static const ParquetVariantProfile disabled;
+ return variant_counters ? *variant_counters : disabled;
+ }
+
RuntimeProfile::Counter* hybrid_selection_batches = nullptr;
RuntimeProfile::Counter* hybrid_selection_ranges = nullptr;
RuntimeProfile::Counter* hybrid_selection_null_fallback_batches = nullptr;
@@ -293,4 +299,23 @@ struct ParquetProfile {
RuntimeProfile::Counter* bloom_filter_read_time = nullptr;
};
+// One immutable counter set per scanner RuntimeProfile. Readers copy its
handles instead of
+// registering every counter again for each file. Lazy initialization keeps
other formats' profiles
+// free of unused Parquet counters.
+class ParquetProfileCache {
+public:
+ explicit ParquetProfileCache(RuntimeProfile* profile) : _profile(profile)
{}
+ const ParquetProfile& get();
+ // Nested table readers share a child RuntimeProfile across scanner
instances. Their counters
+ // must stay on that child, while the root scan owns the cache until every
scanner is closed.
+ ParquetProfileCache* get_child_cache(RuntimeProfile* child_profile);
+
+private:
+ RuntimeProfile* _profile;
+ std::once_flag _initialized;
+ ParquetProfile _counters;
+ std::mutex _children_mutex;
+ std::unordered_map<RuntimeProfile*, std::unique_ptr<ParquetProfileCache>>
_children;
+};
+
} // namespace doris::format::parquet
diff --git a/be/src/format_v2/parquet/parquet_reader.cpp
b/be/src/format_v2/parquet/parquet_reader.cpp
index 0379327486c..c98e794a3c2 100644
--- a/be/src/format_v2/parquet/parquet_reader.cpp
+++ b/be/src/format_v2/parquet/parquet_reader.cpp
@@ -684,8 +684,9 @@
ParquetReader::ParquetReader(std::shared_ptr<io::FileSystemProperties>& system_p
std::shared_ptr<const FileContext> file_context,
int64_t format_split_id, int64_t
format_split_id_end,
std::optional<std::string> hive_parquet_time_zone,
- bool preserve_binary_uuid)
+ bool preserve_binary_uuid, ParquetProfileCache*
profile_cache)
: FileReader(system_properties, file_description, io_ctx, profile),
+ _profile_cache(profile_cache),
_global_rowid_context(global_rowid_context),
_enable_mapping_timestamp_tz(enable_mapping_timestamp_tz),
_enable_mapping_varbinary(enable_mapping_varbinary),
@@ -1373,7 +1374,12 @@ Status ParquetReader::close() {
}
void ParquetReader::_init_profile() {
- _parquet_profile.init(_profile);
+ if (_profile_cache != nullptr) {
+ _parquet_profile = _profile_cache->get();
+ } else {
+ // Standalone readers do not have a scan instance to own a shared
counter set.
+ _parquet_profile.init(_profile);
+ }
}
} // namespace doris::format::parquet
diff --git a/be/src/format_v2/parquet/parquet_reader.h
b/be/src/format_v2/parquet/parquet_reader.h
index d5830b91e6a..a5f2e65d2de 100644
--- a/be/src/format_v2/parquet/parquet_reader.h
+++ b/be/src/format_v2/parquet/parquet_reader.h
@@ -51,7 +51,7 @@ public:
std::shared_ptr<const FileContext> file_context = nullptr,
int64_t format_split_id = -1, int64_t format_split_id_end =
-1,
std::optional<std::string> hive_parquet_time_zone =
std::nullopt,
- bool preserve_binary_uuid = false);
+ bool preserve_binary_uuid = false, ParquetProfileCache*
profile_cache = nullptr);
~ParquetReader() override;
Status init(RuntimeState* state) override;
@@ -100,6 +100,7 @@ private:
std::unique_ptr<ParquetReaderScanState>
_state; // complete scan state (file_context +
schema + scheduler)
ParquetProfile _parquet_profile; // RuntimeProfile counter set
+ ParquetProfileCache* _profile_cache;
ParquetPageCacheStats _reported_page_cache_stats;
std::optional<format::GlobalRowIdContext> _global_rowid_context; // global
RowId context
size_t _batch_size = ParquetScanScheduler::DEFAULT_READ_BATCH_SIZE;
diff --git a/be/src/format_v2/parquet/reader/variant_column_reader.cpp
b/be/src/format_v2/parquet/reader/variant_column_reader.cpp
index 29b8d2639a8..19870b5d438 100644
--- a/be/src/format_v2/parquet/reader/variant_column_reader.cpp
+++ b/be/src/format_v2/parquet/reader/variant_column_reader.cpp
@@ -587,7 +587,8 @@ ColumnVariantV2::MutablePtr encode_variant_column(
int64_t imported_bytes = 0;
RuntimeProfile::Counter* direct_import_time =
profile == nullptr ? nullptr
- :
profile->variant_unshredded_direct_import_time.get();
+ : profile->variant_profile()
+
.variant_unshredded_direct_import_time.get();
{
// Update the shared timer on every exit, including validation
failures. Rows and
// bytes are published only after the complete chunk has been
appended.
@@ -599,13 +600,15 @@ ColumnVariantV2::MutablePtr encode_variant_column(
}
if (profile != nullptr) {
const auto imported_rows = static_cast<int64_t>(end - begin);
- if (profile->variant_unshredded_direct_import_rows != nullptr)
{
-
COUNTER_UPDATE(profile->variant_unshredded_direct_import_rows.get(),
- imported_rows);
+ if
(profile->variant_profile().variant_unshredded_direct_import_rows != nullptr) {
+ COUNTER_UPDATE(
+
profile->variant_profile().variant_unshredded_direct_import_rows.get(),
+ imported_rows);
}
- if (profile->variant_unshredded_direct_import_bytes !=
nullptr) {
-
COUNTER_UPDATE(profile->variant_unshredded_direct_import_bytes.get(),
- imported_bytes);
+ if
(profile->variant_profile().variant_unshredded_direct_import_bytes != nullptr) {
+ COUNTER_UPDATE(
+
profile->variant_profile().variant_unshredded_direct_import_bytes.get(),
+ imported_bytes);
}
}
} else {
@@ -1675,7 +1678,7 @@ public:
std::optional<VariantShreddedTypedValue> find_typed_value(
std::span<const VariantShreddedPathSegment> path) const override {
auto path_miss = [&]() -> std::optional<VariantShreddedTypedValue> {
- update_counter(_profile.variant_direct_leaf_path_misses, 1);
+
update_counter(_profile.variant_profile().variant_direct_leaf_path_misses, 1);
return std::nullopt;
};
if (path.empty()) {
@@ -1745,7 +1748,8 @@ public:
!levels.back().residual_view.present()) {
return path_miss();
}
- update_counter(_profile.variant_residual_seek_rows,
static_cast<int64_t>(rows));
+
update_counter(_profile.variant_profile().variant_residual_seek_rows,
+ static_cast<int64_t>(rows));
return VariantShreddedTypedValue {
.column = nullptr,
.type = nullptr,
@@ -1789,7 +1793,8 @@ public:
if (typed_schema->kind != ParquetColumnSchemaKind::PRIMITIVE ||
check_and_get_column<ColumnNullable>(*typed) == nullptr) {
static_cast<void>(validate_levels());
-
update_counter(_profile.variant_direct_leaf_unsupported_fallbacks, 1);
+
update_counter(_profile.variant_profile().variant_direct_leaf_unsupported_fallbacks,
+ 1);
if (!_complete) {
// Binary element_at evaluates complex prefixes before the
validated leaf.
// Serialize only retained descendants so projected-out
fields stay hidden.
@@ -1806,7 +1811,8 @@ public:
// Decoding a residual needs the root dictionary. A state that
projected the
// metadata away cannot serve those rows, so it keeps the
complete fallback.
if (!metadata.present()) {
-
update_counter(_profile.variant_direct_leaf_residual_fallbacks, 1);
+ update_counter(
+
_profile.variant_profile().variant_direct_leaf_residual_fallbacks, 1);
if (!_complete) {
// A projected state cannot rebuild its root, so
serialize the retained
// descendants instead of demanding the complete
Variant.
@@ -1817,8 +1823,9 @@ public:
}
return std::nullopt;
}
- update_counter(_profile.variant_direct_leaf_rows,
static_cast<int64_t>(rows));
-
update_counter(_profile.variant_direct_leaf_residual_merged_rows,
+
update_counter(_profile.variant_profile().variant_direct_leaf_rows,
+ static_cast<int64_t>(rows));
+
update_counter(_profile.variant_profile().variant_direct_leaf_residual_merged_rows,
static_cast<int64_t>(rows));
return VariantShreddedTypedValue {
.column = nullptr,
@@ -1827,7 +1834,8 @@ public:
levels, path,
metadata_index)};
}
- update_counter(_profile.variant_direct_leaf_rows,
static_cast<int64_t>(rows));
+ update_counter(_profile.variant_profile().variant_direct_leaf_rows,
+ static_cast<int64_t>(rows));
if (!supports_direct_typed_variant_state(*typed_schema)) {
// ColumnVariantV2 typed state carries only a Doris type, so
binary/UUID
// annotations and temporal units would be lost. Normalizing
the leaf through its
@@ -1917,21 +1925,24 @@ public:
const auto borrowed =
borrow_unshredded_path(_unshredded_prefix);
int64_t copied_bytes = 0;
{
-
SCOPED_TIMER(_profile.variant_unshredded_direct_seek_time.get());
+ SCOPED_TIMER(
+
_profile.variant_profile().variant_unshredded_direct_seek_time.get());
_normalized_prefix = normalize_unshredded_path(
*_schema, *_physical, *child_indices, borrowed,
&copied_bytes);
}
const auto rows = static_cast<int64_t>(_physical->size());
- update_counter(_profile.variant_unshredded_direct_seek_rows,
rows);
- update_counter(_profile.variant_unshredded_direct_seek_bytes,
copied_bytes);
+
update_counter(_profile.variant_profile().variant_unshredded_direct_seek_rows,
+ rows);
+
update_counter(_profile.variant_profile().variant_unshredded_direct_seek_bytes,
+ copied_bytes);
}
const auto& nullable = assert_cast<const
ColumnNullable&>(*_normalized_prefix);
return assert_cast<const
ColumnVariantV2&>(nullable.get_nested_column());
}
if (!_materialized) {
- SCOPED_TIMER(_profile.variant_reconstruction_time.get());
+
SCOPED_TIMER(_profile.variant_profile().variant_reconstruction_time.get());
_materialized = encode_variant_column(*_schema, *_physical, true,
&_profile);
- update_counter(_profile.variant_reconstructed_rows,
+
update_counter(_profile.variant_profile().variant_reconstructed_rows,
static_cast<int64_t>(_physical->size()));
}
return *_materialized;
@@ -2062,7 +2073,7 @@ private:
std::shared_ptr<const UnshreddedMetadataIndex> metadata_index;
UnshreddedPathScan scan;
{
- SCOPED_TIMER(_profile.variant_unshredded_direct_seek_time.get());
+
SCOPED_TIMER(_profile.variant_profile().variant_unshredded_direct_seek_time.get());
metadata_index = get_unshredded_metadata_index(*child_indices);
scan = scan_unshredded_path(
*_schema, *_physical, *child_indices, *metadata_index,
@@ -2073,22 +2084,23 @@ private:
}
const auto rows = static_cast<int64_t>(_physical->size());
- update_counter(_profile.variant_unshredded_direct_seek_rows, rows);
+
update_counter(_profile.variant_profile().variant_unshredded_direct_seek_rows,
rows);
if (_unshredded_path_cache) {
- update_counter(_profile.variant_unshredded_prefix_reuse_rows,
rows);
+
update_counter(_profile.variant_profile().variant_unshredded_prefix_reuse_rows,
rows);
}
MutableColumnPtr values;
std::shared_ptr<ParquetVariantShreddedState> subtree_state;
if (scan.typed_values) {
values =
ColumnVariantV2::create_typed(std::move(scan.typed_values),
std::move(scan.typed_type));
- update_counter(_profile.variant_unshredded_direct_seek_bytes,
scan.copied_bytes);
- update_counter(_profile.variant_direct_leaf_rows, rows);
+
update_counter(_profile.variant_profile().variant_unshredded_direct_seek_bytes,
+ scan.copied_bytes);
+
update_counter(_profile.variant_profile().variant_direct_leaf_rows, rows);
} else {
subtree_state = std::make_shared<ParquetVariantShreddedState>(
_schema, _physical, _complete, _profile, combined,
std::move(scan.path_cache));
values = ColumnVariantV2::create_shredded(subtree_state);
- update_counter(_profile.variant_direct_subtree_rows, rows);
+
update_counter(_profile.variant_profile().variant_direct_subtree_rows, rows);
}
ColumnPtr result = ColumnNullable::create(std::move(values),
std::move(scan.outer_nulls));
return result;
diff --git a/be/src/format_v2/table/hudi_reader.cpp
b/be/src/format_v2/table/hudi_reader.cpp
index 88fdc484695..a9ce03f7536 100644
--- a/be/src/format_v2/table/hudi_reader.cpp
+++ b/be/src/format_v2/table/hudi_reader.cpp
@@ -204,6 +204,7 @@ Status
HudiHybridReader::_init_child_reader(format::TableReader* reader,
.io_ctx = _io_ctx,
.runtime_state = _runtime_state,
.scanner_profile = _scanner_profile,
+ .parquet_profile_cache = _parquet_profile_cache,
.push_down_agg_type = _push_down_agg_type,
.push_down_count_columns = _push_down_count_columns,
.condition_cache_digest = _condition_cache_digest,
diff --git
a/be/src/format_v2/table/iceberg_position_delete_sys_table_reader.cpp
b/be/src/format_v2/table/iceberg_position_delete_sys_table_reader.cpp
index cd84a255e69..03a4a69057d 100644
--- a/be/src/format_v2/table/iceberg_position_delete_sys_table_reader.cpp
+++ b/be/src/format_v2/table/iceberg_position_delete_sys_table_reader.cpp
@@ -35,6 +35,7 @@
#include "core/types.h"
#include "format/table/iceberg_delete_file_reader_helper.h"
#include "format/table/parquet_utils.h"
+#include "format_v2/parquet/parquet_profile.h"
#include "format_v2/table/iceberg_reader.h"
#include "format_v2/table/iceberg_schema_utils.h"
#include "runtime/descriptors.h"
@@ -323,6 +324,15 @@ Status
IcebergPositionDeleteSysTableV2Reader::_init_position_delete_reader() {
// readers share the scanner profile and initialize concurrently, so
lookup and creation
// must also be atomic to preserve the unique child-name invariant.
_position_reader_profile =
_scanner_profile->get_or_create_child(kPositionReaderProfile);
+ if (_parquet_profile_cache != nullptr) {
+ _position_profile_cache =
+
_parquet_profile_cache->get_child_cache(_position_reader_profile);
+ } else {
+ // Standalone callers have no scan-owned cache. Keep one across
this reader's splits.
+ _standalone_position_profile_cache =
+
std::make_unique<parquet::ParquetProfileCache>(_position_reader_profile);
+ _position_profile_cache = _standalone_position_profile_cache.get();
+ }
}
_position_reader = std::make_unique<PositionDeleteFileTableReader>();
RETURN_IF_ERROR(_position_reader->init({
@@ -333,6 +343,7 @@ Status
IcebergPositionDeleteSysTableV2Reader::_init_position_delete_reader() {
.io_ctx = _io_ctx,
.runtime_state = _runtime_state,
.scanner_profile = _position_reader_profile,
+ .parquet_profile_cache = _position_profile_cache,
.file_slot_descs = nullptr,
.push_down_agg_type = TPushAggOp::type::NONE,
.condition_cache_digest = 0,
diff --git a/be/src/format_v2/table/iceberg_position_delete_sys_table_reader.h
b/be/src/format_v2/table/iceberg_position_delete_sys_table_reader.h
index 09756daaabe..bbd535ed31f 100644
--- a/be/src/format_v2/table/iceberg_position_delete_sys_table_reader.h
+++ b/be/src/format_v2/table/iceberg_position_delete_sys_table_reader.h
@@ -77,6 +77,8 @@ private:
DeleteFileKind _delete_file_kind = DeleteFileKind::POSITION_DELETE;
std::unique_ptr<format::TableReader> _position_reader;
RuntimeProfile* _position_reader_profile = nullptr;
+ std::unique_ptr<parquet::ParquetProfileCache>
_standalone_position_profile_cache;
+ parquet::ParquetProfileCache* _position_profile_cache = nullptr;
std::vector<ReadColumn> _read_columns;
ColumnPtr _partition_value;
roaring::Roaring64Map _dv_positions;
diff --git a/be/src/format_v2/table/iceberg_reader.cpp
b/be/src/format_v2/table/iceberg_reader.cpp
index 88eb497c3dc..03de2fa308c 100644
--- a/be/src/format_v2/table/iceberg_reader.cpp
+++ b/be/src/format_v2/table/iceberg_reader.cpp
@@ -1547,7 +1547,7 @@ Status
IcebergTableReader::_create_delete_file_reader(const TIcebergDeleteFileDe
*reader = std::make_unique<format::parquet::ParquetReader>(
system_properties, file_description, io_ctx, _scanner_profile,
std::nullopt,
enable_mapping_timestamp_tz, enable_mapping_varbinary,
nullptr, -1, -1, "",
- preserve_binary_uuid());
+ preserve_binary_uuid(), _parquet_profile_cache);
} else {
*reader = std::make_unique<format::orc::OrcReader>(
system_properties, file_description, io_ctx, _scanner_profile,
std::nullopt,
diff --git a/be/src/format_v2/table/load_reader.cpp
b/be/src/format_v2/table/load_reader.cpp
index 2f592272045..1e501a3cb03 100644
--- a/be/src/format_v2/table/load_reader.cpp
+++ b/be/src/format_v2/table/load_reader.cpp
@@ -88,6 +88,7 @@ Status LoadTableReader::init(TableReadOptions&& options) {
.io_ctx = options.io_ctx,
.runtime_state = options.runtime_state,
.scanner_profile = options.scanner_profile,
+ .parquet_profile_cache =
options.parquet_profile_cache,
.file_slot_descs = &_source_file_slots,
.push_down_agg_type = TPushAggOp::NONE,
.condition_cache_digest = 0});
diff --git a/be/src/format_v2/table/paimon_reader.cpp
b/be/src/format_v2/table/paimon_reader.cpp
index 2ec6bde53f2..de373b1e5f8 100644
--- a/be/src/format_v2/table/paimon_reader.cpp
+++ b/be/src/format_v2/table/paimon_reader.cpp
@@ -496,6 +496,7 @@ Status
PaimonHybridReader::_init_child_reader(format::TableReader* reader,
.io_ctx = _io_ctx,
.runtime_state = _runtime_state,
.scanner_profile = _scanner_profile,
+ .parquet_profile_cache = _parquet_profile_cache,
.push_down_agg_type = _push_down_agg_type,
.push_down_count_columns = _push_down_count_columns,
.condition_cache_digest = _condition_cache_digest,
diff --git a/be/src/format_v2/table_reader.cpp
b/be/src/format_v2/table_reader.cpp
index 4750aef67f8..9e946ef65c2 100644
--- a/be/src/format_v2/table_reader.cpp
+++ b/be/src/format_v2/table_reader.cpp
@@ -1125,6 +1125,7 @@
TableReader::_find_table_column_identity_path_by_field_id(int32_t field_id,
Status TableReader::init(TableReadOptions&& options) {
_scanner_profile = options.scanner_profile;
+ _parquet_profile_cache = options.parquet_profile_cache;
if (_scanner_profile != nullptr) {
const auto hierarchy =
file_scan_profile::ensure_hierarchy(_scanner_profile);
static const char* table_profile = file_scan_profile::TABLE_READER;
@@ -1693,7 +1694,8 @@ Status
TableReader::create_file_reader(std::unique_ptr<FileReader>* reader) {
_system_properties, _current_task->data_file, _io_ctx,
_scanner_profile,
_global_rowid_context, enable_mapping_timestamp_tz,
enable_mapping_varbinary,
_current_task->file_context, _current_task->format_split_id,
- _current_task->format_split_id_end, hive_parquet_time_zone,
preserve_binary_uuid());
+ _current_task->format_split_id_end, hive_parquet_time_zone,
preserve_binary_uuid(),
+ _parquet_profile_cache);
return Status::OK();
}
if (_format == FileFormat::ORC) {
diff --git a/be/src/format_v2/table_reader.h b/be/src/format_v2/table_reader.h
index b71f1be3c99..1f8a6d9b29a 100644
--- a/be/src/format_v2/table_reader.h
+++ b/be/src/format_v2/table_reader.h
@@ -76,6 +76,10 @@ class RuntimeState;
namespace doris::format {
+namespace parquet {
+class ParquetProfileCache;
+}
+
using DeleteRows = std::vector<int64_t>;
// Row-level predicates on table/global schema. They are rewritten to
file-local expressions when
@@ -149,6 +153,7 @@ struct TableReadOptions {
std::shared_ptr<io::IOContext> io_ctx;
RuntimeState* runtime_state;
RuntimeProfile* scanner_profile;
+ parquet::ParquetProfileCache* parquet_profile_cache = nullptr;
// File formats without complete self-describing metadata, such as CSV,
Text, and JSON, need
// the FE-planned physical file slots to build their file-local schema and
deserialize values.
const std::vector<SlotDescriptor*>* file_slot_descs = nullptr;
@@ -2035,6 +2040,7 @@ protected:
std::shared_ptr<io::IOContext> _io_ctx;
RuntimeState* _runtime_state;
RuntimeProfile* _scanner_profile;
+ parquet::ParquetProfileCache* _parquet_profile_cache = nullptr;
const std::vector<SlotDescriptor*>* _file_slot_descs = nullptr;
FileFormat _format;
TPushAggOp::type _push_down_agg_type = TPushAggOp::type::NONE;
diff --git a/be/test/format_v2/parquet/parquet_profile_test.cpp
b/be/test/format_v2/parquet/parquet_profile_test.cpp
new file mode 100644
index 00000000000..bdb999b5993
--- /dev/null
+++ b/be/test/format_v2/parquet/parquet_profile_test.cpp
@@ -0,0 +1,169 @@
+// 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/parquet_profile.h"
+
+#include <gtest/gtest.h>
+
+#include <array>
+#include <atomic>
+#include <chrono>
+#include <future>
+#include <memory>
+#include <thread>
+#include <vector>
+
+namespace doris::format::parquet {
+
+TEST(ParquetProfileCacheTest, InitializesLazilyAndKeepsCounterIdentity) {
+ RuntimeProfile profile("scan");
+ ParquetProfileCache cache(&profile);
+ EXPECT_EQ(profile.get_counter("ParquetReader"), nullptr);
+ const auto& counters = cache.get();
+ ASSERT_NE(counters.total_time, nullptr);
+ EXPECT_EQ(counters.total_time, profile.get_counter("ParquetReader"));
+ counters.total_time->update(17);
+ EXPECT_EQ(&cache.get(), &counters);
+ EXPECT_EQ(cache.get().total_time->value(), 17);
+}
+
+TEST(ParquetProfileCacheTest, ConcurrentReadersShareOneCounterSet) {
+ RuntimeProfile profile("scan");
+ ParquetProfileCache cache(&profile);
+ std::array<const ParquetProfile*, 16> handles {};
+ std::atomic<bool> start = false;
+ std::vector<std::thread> readers;
+ for (size_t i = 0; i < handles.size(); ++i) {
+ readers.emplace_back([&, i] {
+ while (!start.load(std::memory_order_acquire)) {
+ std::this_thread::yield();
+ }
+ handles[i] = &cache.get();
+ for (int file = 0; file < 100; ++file) {
+ cache.get().open_file_num->update(1);
+ }
+ });
+ }
+ start.store(true, std::memory_order_release);
+ for (auto& reader : readers) {
+ reader.join();
+ }
+ for (const auto* handle : handles) {
+ EXPECT_EQ(handle, handles[0]);
+ }
+ EXPECT_EQ(cache.get().open_file_num->value(), 1600);
+}
+
+TEST(ParquetProfileCacheTest,
ChildCachesShareAcrossThreadsButKeepProfilesSeparate) {
+ RuntimeProfile root("scan");
+ RuntimeProfile* child = root.get_or_create_child("nested");
+ RuntimeProfile* other = root.get_or_create_child("other");
+ ParquetProfileCache cache(&root);
+ std::array<ParquetProfileCache*, 16> children {};
+ std::vector<std::thread> readers;
+ for (size_t i = 0; i < children.size(); ++i) {
+ readers.emplace_back([&, i] { children[i] =
cache.get_child_cache(child); });
+ }
+ for (auto& reader : readers) {
+ reader.join();
+ }
+ for (auto* handle : children) {
+ EXPECT_EQ(handle, children[0]);
+ }
+ EXPECT_NE(cache.get_child_cache(other), children[0]);
+ EXPECT_NE(cache.get().total_time, children[0]->get().total_time);
+ EXPECT_EQ(child->get_counter("ParquetReader"),
children[0]->get().total_time);
+}
+
+TEST(ParquetProfileCacheTest,
CachedReadersDoNotAcquireCounterRegistrationLock) {
+ RuntimeProfile profile("scan");
+ ParquetProfileCache cache(&profile);
+ const auto* counters = &cache.get();
+ // BE tests compile with -fno-access-control, as do the existing
RuntimeProfile tests.
+ std::unique_lock registration_lock(profile._counter_map_lock);
+ std::promise<const ParquetProfile*> result;
+ auto future = result.get_future();
+ std::thread reader([&] { result.set_value(&cache.get()); });
+ const auto status = future.wait_for(std::chrono::seconds(5));
+ registration_lock.unlock();
+ reader.join();
+ EXPECT_EQ(status, std::future_status::ready);
+ EXPECT_EQ(future.get(), counters);
+}
+
+TEST(ParquetProfileCacheTest, ScanInstancesKeepIndependentCounters) {
+ RuntimeProfile first("first");
+ RuntimeProfile second("second");
+ ParquetProfileCache first_cache(&first);
+ ParquetProfileCache second_cache(&second);
+ first_cache.get().raw_rows_read->update(3);
+ EXPECT_NE(first_cache.get().raw_rows_read,
second_cache.get().raw_rows_read);
+ EXPECT_EQ(second_cache.get().raw_rows_read->value(), 0);
+}
+
+TEST(ParquetProfileCacheTest, NullProfileKeepsCountersDisabled) {
+ ParquetProfileCache cache(nullptr);
+ EXPECT_EQ(cache.get().total_time, nullptr);
+ EXPECT_EQ(cache.get().variant_profile().variant_reconstruction_time,
nullptr);
+ EXPECT_EQ(&cache.get(), &cache.get());
+}
+
+TEST(ParquetProfileCacheTest, VariantCountersOutliveScanProfile) {
+ ParquetProfile reader_counters;
+ {
+ RuntimeProfile profile("scan");
+ ParquetProfileCache cache(&profile);
+ reader_counters = cache.get();
+ }
+ // Deferred Variant materialization retains shared counters after the
scanner is destroyed.
+ ASSERT_NE(reader_counters.variant_profile().variant_reconstruction_time,
nullptr);
+ reader_counters.variant_profile().variant_reconstruction_time->update(23);
+
EXPECT_EQ(reader_counters.variant_profile().variant_reconstruction_time->value(),
23);
+ reader_counters.variant_profile().variant_direct_leaf_rows->update(7);
+
EXPECT_EQ(reader_counters.variant_profile().variant_direct_leaf_rows->value(),
7);
+}
+
+TEST(ParquetProfileCacheTest,
ColumnProfileCopiesRetainOneOwnerAfterScanDestruction) {
+ ParquetColumnReaderProfile reader;
+ std::weak_ptr<const ParquetVariantProfile> lifetime;
+ {
+ RuntimeProfile profile("scan");
+ ParquetProfileCache cache(&profile);
+ reader = cache.get().column_reader_profile();
+ lifetime = reader.variant_counters;
+ const auto& counter =
reader.variant_profile().variant_unshredded_direct_import_rows;
+ const auto references = counter.use_count();
+ auto copy = reader;
+ EXPECT_EQ(copy.variant_counters, reader.variant_counters);
+ EXPECT_EQ(counter.use_count(), references);
+
copy.variant_profile().variant_unshredded_direct_import_rows->update(9);
+ }
+ EXPECT_FALSE(lifetime.expired());
+
EXPECT_EQ(reader.variant_profile().variant_unshredded_direct_import_rows->value(),
9);
+ reader.variant_profile().variant_unshredded_direct_import_rows->update(3);
+
EXPECT_EQ(reader.variant_profile().variant_unshredded_direct_import_rows->value(),
12);
+ reader = {};
+ EXPECT_TRUE(lifetime.expired());
+}
+
+TEST(ParquetProfileCacheTest, DefaultColumnProfileHasDisabledVariantCounters) {
+ ParquetColumnReaderProfile reader;
+ EXPECT_EQ(reader.variant_profile().variant_unshredded_direct_import_time,
nullptr);
+ EXPECT_EQ(reader.variant_profile().variant_reconstruction_time, nullptr);
+}
+
+} // namespace doris::format::parquet
diff --git a/be/test/format_v2/parquet/parquet_reader_test.cpp
b/be/test/format_v2/parquet/parquet_reader_test.cpp
index ae195250a1d..3d036c26dce 100644
--- a/be/test/format_v2/parquet/parquet_reader_test.cpp
+++ b/be/test/format_v2/parquet/parquet_reader_test.cpp
@@ -2327,6 +2327,56 @@ TEST_F(NewParquetReaderTest,
UuidPlainDictionaryNullableAndMappingMatrix) {
}
}
+TEST_F(NewParquetReaderTest, InitializesProfileOnlyOncePerFile) {
+ class CountingParquetReader final : public format::parquet::ParquetReader {
+ public:
+ using ParquetReader::ParquetReader;
+ int profile_initializations = 0;
+
+ protected:
+ void _init_profile() override {
+ ++profile_initializations;
+ ParquetReader::_init_profile();
+ }
+ };
+ auto properties = std::make_shared<io::FileSystemProperties>();
+ properties->system_type = TFileType::FILE_LOCAL;
+ auto file = std::make_unique<io::FileDescription>();
+ file->path = _file_path;
+ file->file_size =
static_cast<int64_t>(std::filesystem::file_size(_file_path));
+ RuntimeProfile profile("scan");
+ CountingParquetReader reader(properties, file, nullptr, &profile);
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ ASSERT_TRUE(reader.init(&state).ok());
+ EXPECT_EQ(reader.profile_initializations, 1);
+ std::vector<format::ColumnDefinition> schema;
+ ASSERT_TRUE(reader.get_schema(&schema).ok());
+ EXPECT_EQ(schema.size(), 2);
+ EXPECT_TRUE(reader.close().ok());
+}
+
+TEST_F(NewParquetReaderTest, ReusesProfileHandlesAcrossFiles) {
+ RuntimeProfile profile("scan");
+ format::parquet::ParquetProfileCache cache(&profile);
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ for (int file_index = 0; file_index < 3; ++file_index) {
+ auto properties = std::make_shared<io::FileSystemProperties>();
+ properties->system_type = TFileType::FILE_LOCAL;
+ auto file = std::make_unique<io::FileDescription>();
+ file->path = _file_path;
+ file->file_size =
static_cast<int64_t>(std::filesystem::file_size(_file_path));
+ format::parquet::ParquetReader reader(properties, file, nullptr,
&profile, std::nullopt,
+ false, false, nullptr, -1, -1,
std::nullopt, false,
+ &cache);
+ ASSERT_TRUE(reader.init(&state).ok());
+ std::vector<format::ColumnDefinition> schema;
+ ASSERT_TRUE(reader.get_schema(&schema).ok());
+ EXPECT_EQ(schema.size(), 2);
+ ASSERT_TRUE(reader.close().ok());
+ EXPECT_EQ(cache.get().open_file_num->value(), file_index + 1);
+ }
+}
+
TEST_F(NewParquetReaderTest, GetSchemaReturnsFileLocalColumns) {
auto reader = create_reader();
RuntimeState state {TQueryOptions(), TQueryGlobals()};
diff --git a/be/test/format_v2/parquet/variant_column_reader_test.cpp
b/be/test/format_v2/parquet/variant_column_reader_test.cpp
index 0d179f5e5a7..b8423c1f74c 100644
--- a/be/test/format_v2/parquet/variant_column_reader_test.cpp
+++ b/be/test/format_v2/parquet/variant_column_reader_test.cpp
@@ -2553,10 +2553,10 @@ TEST(VariantColumnReaderTest,
ShreddedStateOutlivesScannerProfile) {
parquet_profile.init(runtime_profile.get());
ParquetProfile reused_profile;
reused_profile.init(runtime_profile.get());
- EXPECT_EQ(parquet_profile.variant_reconstructed_rows,
- reused_profile.variant_reconstructed_rows);
- EXPECT_EQ(parquet_profile.variant_unshredded_direct_import_rows,
- reused_profile.variant_unshredded_direct_import_rows);
+ EXPECT_EQ(parquet_profile.variant_profile().variant_reconstructed_rows,
+ reused_profile.variant_profile().variant_reconstructed_rows);
+
EXPECT_EQ(parquet_profile.variant_profile().variant_unshredded_direct_import_rows,
+
reused_profile.variant_profile().variant_unshredded_direct_import_rows);
auto visible_output =
make_nullable(std::make_shared<DataTypeVariantV2>())->create_column();
ASSERT_TRUE(materialize_variant_rows(shredded_int64_schema(),
shredded_int64_physical({7}),
diff --git a/be/test/format_v2/table/hudi_reader_test.cpp
b/be/test/format_v2/table/hudi_reader_test.cpp
index af72dcf1857..c2984430560 100644
--- a/be/test/format_v2/table/hudi_reader_test.cpp
+++ b/be/test/format_v2/table/hudi_reader_test.cpp
@@ -38,6 +38,7 @@
#include "core/data_type/data_type_struct.h"
#include "core/field.h"
#include "format_v2/column_data.h"
+#include "format_v2/parquet/parquet_reader.h"
#include "gen_cpp/ExternalTableSchema_types.h"
#include "gen_cpp/PlanNodes_types.h"
#include "io/io_common.h"
@@ -156,9 +157,51 @@ public:
int refresh_count = 0;
};
+class ProfileCacheTrackingTableReader final : public TableReader {
+public:
+ Status init(TableReadOptions&& options) override {
+ observed_cache = options.parquet_profile_cache;
+ return TableReader::init(std::move(options));
+ }
+
+ Status prepare_split(const SplitReadOptions&) override { return
Status::OK(); }
+
+ parquet::ParquetProfileCache* observed_cache = nullptr;
+};
+
// Scenario: FileScannerV2 Hudi native reader uses the split schema id to
annotate the physical
// file schema before TableColumnMapper runs. This keeps schema-evolved Hudi
files on field-id
// mapping, including renamed nested children.
+TEST(HudiReaderTest, CustomParquetFactoryReusesScannerCounters) {
+ RuntimeProfile profile("hudi");
+ parquet::ParquetProfileCache cache(&profile);
+ TQueryOptions options;
+ TQueryGlobals globals;
+ RuntimeState state(options, globals);
+ TFileScanRangeParams scan_params;
+ hudi::HudiReader reader;
+ ASSERT_TRUE(reader.init({.projected_columns = {},
+ .conjuncts = {},
+ .format = FileFormat::PARQUET,
+ .scan_params = &scan_params,
+ .io_ctx = nullptr,
+ .runtime_state = &state,
+ .scanner_profile = &profile,
+ .parquet_profile_cache = &cache})
+ .ok());
+ reader._current_task = std::make_unique<ScanTask>();
+ for (int i = 0; i < 2; ++i) {
+ reader._current_task->data_file =
std::make_unique<io::FileDescription>();
+ std::unique_ptr<FileReader> file;
+ ASSERT_TRUE(reader.create_file_reader(&file).ok());
+ auto* parquet = dynamic_cast<parquet::ParquetReader*>(file.get());
+ ASSERT_NE(parquet, nullptr);
+ EXPECT_EQ(parquet->_profile_cache, &cache);
+ parquet->_init_profile();
+ EXPECT_EQ(parquet->_parquet_profile.total_time,
cache.get().total_time);
+ }
+}
+
TEST(HudiReaderTest, AnnotatesFileSchemaFromSplitHistorySchema) {
TFileScanRangeParams scan_params;
scan_params.__set_current_schema_id(200);
@@ -277,6 +320,37 @@ TEST(HudiHybridReaderTest,
AdaptiveBatchSizeReachesBothChildReaders) {
EXPECT_EQ(child_batch_sizes.second, 123);
}
+TEST(HudiHybridReaderTest, ForwardsParquetProfileCacheToNativeChild) {
+ RuntimeProfile profile("hudi_hybrid");
+ parquet::ParquetProfileCache cache(&profile);
+ TFileScanRangeParams scan_params;
+ scan_params.__set_format_type(TFileFormatType::FORMAT_PARQUET);
+ hudi::HudiHybridReader reader;
+ ProfileCacheTrackingTableReader* native = nullptr;
+ reader.TEST_set_child_reader_factories(
+ [&] {
+ auto child =
std::make_unique<ProfileCacheTrackingTableReader>();
+ native = child.get();
+ return child;
+ },
+ [] { return std::make_unique<TableReader>(); });
+ ASSERT_TRUE(reader.init({.projected_columns = {},
+ .conjuncts = {},
+ .format = FileFormat::PARQUET,
+ .scan_params = &scan_params,
+ .io_ctx = nullptr,
+ .runtime_state = nullptr,
+ .scanner_profile = &profile,
+ .parquet_profile_cache = &cache})
+ .ok());
+ SplitReadOptions split;
+ split.current_split_format = FileFormat::PARQUET;
+ split.current_range.__set_format_type(TFileFormatType::FORMAT_PARQUET);
+ ASSERT_TRUE(reader.prepare_split(split).ok());
+ ASSERT_NE(native, nullptr);
+ EXPECT_EQ(native->observed_cache, &cache);
+}
+
TEST(HudiHybridReaderTest, AggregatesConditionCacheHitsFromBothChildren) {
hudi::HudiHybridReader reader;
reader.TEST_install_batch_size_children();
diff --git
a/be/test/format_v2/table/iceberg_position_delete_sys_table_reader_test.cpp
b/be/test/format_v2/table/iceberg_position_delete_sys_table_reader_test.cpp
index 8cf538eb0cf..2fde7a6fba6 100644
--- a/be/test/format_v2/table/iceberg_position_delete_sys_table_reader_test.cpp
+++ b/be/test/format_v2/table/iceberg_position_delete_sys_table_reader_test.cpp
@@ -22,6 +22,8 @@
#include "core/data_type/data_type_number.h"
#include "core/data_type/data_type_struct.h"
#include "format/table/iceberg_scan_semantics.h"
+#include "format_v2/parquet/parquet_profile.h"
+#include "format_v2/parquet/parquet_reader.h"
#include "runtime/runtime_profile.h"
#include "runtime/runtime_state.h"
@@ -62,6 +64,49 @@ TEST(IcebergPositionDeleteSysTableV2ProfileTest,
UsesDistinctProfileForNestedPos
reader._position_reader_profile);
}
+TEST(IcebergPositionDeleteSysTableV2ProfileTest,
ReusesChildCountersAcrossSplitsAndScanners) {
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ RuntimeProfile profile("position_delete_system_table_profile");
+ parquet::ParquetProfileCache outer_cache(&profile);
+ TFileScanRangeParams params;
+ std::vector<SlotDescriptor*> file_slot_descs;
+ TIcebergDeleteFileDesc delete_file;
+ delete_file.__set_content(1);
+ delete_file.__set_file_format(TFileFormatType::FORMAT_PARQUET);
+ delete_file.__set_path("/not-opened-during-prepare.parquet");
+ parquet::ParquetProfileCache* shared_child = nullptr;
+ for (int scanner = 0; scanner < 2; ++scanner) {
+ IcebergPositionDeleteSysTableV2Reader reader;
+ reader._runtime_state = &state;
+ reader._scanner_profile = &profile;
+ reader._parquet_profile_cache = &outer_cache;
+ reader._scan_params = ¶ms;
+ reader._file_slot_descs = &file_slot_descs;
+ for (int split = 0; split < 2; ++split) {
+ reader._current_range = range_with_delete_file(delete_file);
+ ASSERT_TRUE(reader._init_split().ok());
+ auto* cache = reader._position_reader->_parquet_profile_cache;
+ ASSERT_NE(cache, nullptr);
+ ASSERT_NE(cache, &outer_cache);
+ if (shared_child == nullptr) {
+ shared_child = cache;
+ }
+ EXPECT_EQ(cache, shared_child);
+ std::unique_ptr<format::FileReader> file;
+
ASSERT_TRUE(reader._position_reader->create_file_reader(&file).ok());
+ auto* parquet_reader =
dynamic_cast<parquet::ParquetReader*>(file.get());
+ ASSERT_NE(parquet_reader, nullptr);
+ EXPECT_EQ(parquet_reader->_profile_cache, shared_child);
+ parquet_reader->_init_profile();
+ EXPECT_EQ(parquet_reader->_parquet_profile.total_time,
shared_child->get().total_time);
+ }
+ }
+ EXPECT_EQ(profile.get_counter("FileFooterReadCalls"), nullptr);
+ EXPECT_NE(profile.get_child("IcebergPositionDeleteFileReader")
+ ->get_counter("FileFooterReadCalls"),
+ nullptr);
+}
+
TEST(IcebergPositionDeleteSysTableV2ProfileTest,
PreparesNestedRowInitialDefaults) {
IcebergPositionDeleteSysTableV2Reader reader;
TFileScanRangeParams params;
diff --git a/be/test/format_v2/table/iceberg_reader_test.cpp
b/be/test/format_v2/table/iceberg_reader_test.cpp
index b8193879e4c..2b7a5fad5dd 100644
--- a/be/test/format_v2/table/iceberg_reader_test.cpp
+++ b/be/test/format_v2/table/iceberg_reader_test.cpp
@@ -75,6 +75,7 @@
#include "format/format_common.h"
#include "format/orc/orc_memory_stream_test.h"
#include "format/table/deletion_vector_reader.h"
+#include "format_v2/parquet/parquet_reader.h"
#include "format_v2/table_reader.h"
#include "gen_cpp/Exprs_types.h"
#include "gen_cpp/ExternalTableSchema_types.h"
@@ -5971,5 +5972,41 @@ TEST(IcebergV2ReaderTest,
IcebergEqualityDeleteFileSizeUnknownFallsBackToStat) {
std::filesystem::remove_all(test_dir);
}
+TEST(IcebergTableReaderTest, DeleteParquetFactoryReusesScannerCounters) {
+ RuntimeProfile profile("iceberg");
+ doris::format::parquet::ParquetProfileCache cache(&profile);
+ doris::format::iceberg::IcebergTableReader reader;
+ reader._scanner_profile = &profile;
+ reader._parquet_profile_cache = &cache;
+ TFileScanRangeParams params;
+ TQueryOptions options;
+ TQueryGlobals globals;
+ RuntimeState state(options, globals);
+ reader._runtime_state = &state;
+ IcebergDeleteFileIOContext io_context(&state);
+ const auto file_path =
+ (std::filesystem::temp_directory_path() /
+ ("iceberg_profile_" +
+
std::to_string(std::chrono::steady_clock::now().time_since_epoch().count()) +
+ ".parquet"))
+ .string();
+ write_single_int_parquet_file(file_path, "id", {1}, 1);
+ for (int i = 0; i < 2; ++i) {
+ TIcebergDeleteFileDesc desc;
+ desc.__set_file_format(TFileFormatType::FORMAT_PARQUET);
+ desc.__set_path(file_path);
+ desc.__set_file_size(std::filesystem::file_size(file_path));
+ std::unique_ptr<doris::format::FileReader> file;
+ const auto status = reader._create_delete_file_reader(desc, params,
&io_context, &file);
+ ASSERT_TRUE(status.ok()) << status.to_string();
+ auto* parquet =
dynamic_cast<doris::format::parquet::ParquetReader*>(file.get());
+ ASSERT_NE(parquet, nullptr);
+ EXPECT_EQ(parquet->_profile_cache, &cache);
+ parquet->_init_profile();
+ EXPECT_EQ(parquet->_parquet_profile.total_time,
cache.get().total_time);
+ }
+ std::filesystem::remove(file_path);
+}
+
} // namespace
} // namespace doris::format
diff --git a/be/test/format_v2/table/load_reader_test.cpp
b/be/test/format_v2/table/load_reader_test.cpp
index a88931bd4d6..6b1c3c399d8 100644
--- a/be/test/format_v2/table/load_reader_test.cpp
+++ b/be/test/format_v2/table/load_reader_test.cpp
@@ -30,6 +30,7 @@
#include "core/data_type/data_type_nullable.h"
#include "core/data_type/data_type_number.h"
#include "core/data_type/data_type_string.h"
+#include "format_v2/parquet/parquet_profile.h"
#include "io/fs/kafka_consumer_pipe.h"
#include "load/stream_load/new_load_stream_mgr.h"
#include "load/stream_load/stream_load_context.h"
@@ -130,6 +131,7 @@ public:
TestLoadTableReader(std::unique_ptr<SourceReader> source_reader,
TupleDescriptor* source,
TupleDescriptor* destination)
: LoadTableReader(source, destination),
_input(std::move(source_reader)) {}
+ parquet::ParquetProfileCache* parquet_profile_cache() const { return
_parquet_profile_cache; }
Status init(TableReadOptions&& options) override {
RETURN_IF_ERROR(LoadTableReader::init(std::move(options)));
_input->columns = _projected_columns;
@@ -236,7 +238,8 @@ protected:
cast.__set_num_children(1);
params.expr_of_dest_slot[20].nodes.insert(params.expr_of_dest_slot[20].nodes.begin(),
cast);
}
- Status init(LoadTableReader& reader, FileFormat format = FileFormat::CSV) {
+ Status init(LoadTableReader& reader, FileFormat format = FileFormat::CSV,
+ parquet::ParquetProfileCache* parquet_profile_cache = nullptr)
{
TTupleDescriptor tuple;
tuple.__set_id(0);
tuple.__set_byteSize(0);
@@ -259,7 +262,8 @@ protected:
.scan_params = ¶ms,
.io_ctx = io,
.runtime_state = &state,
- .scanner_profile = &profile});
+ .scanner_profile = &profile,
+ .parquet_profile_cache = parquet_profile_cache});
}
void add_partial_update_columns(bool key) {
TTupleDescriptor tuple;
@@ -340,6 +344,15 @@ protected:
std::unique_ptr<SlotDescriptor> source_slot, destination_slot;
};
+TEST_F(LoadReaderTest, ForwardsParquetProfileCacheToBaseReader) {
+ parquet::ParquetProfileCache cache(&profile);
+ TestLoadTableReader reader(std::make_unique<SourceReader>(), source.get(),
destination.get());
+ const auto status = init(reader, FileFormat::PARQUET, &cache);
+ ASSERT_TRUE(status.ok()) << status;
+ EXPECT_EQ(reader.parquet_profile_cache(), &cache);
+ ASSERT_TRUE(reader.close().ok());
+}
+
class LoadJsonPipeTest : public LoadReaderTest {
protected:
void SetUp() override {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
