This is an automated email from the ASF dual-hosted git repository.
Gabriel39 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 7073c51342d [improvement](be) Reuse Parquet profile counters across
external scan files (#69300)
7073c51342d is described below
commit 7073c51342df12b44938dde722303a7229a32cdb
Author: 924060929 <[email protected]>
AuthorDate: Tue Oct 20 16:48:28 2026 +0800
[improvement](be) Reuse Parquet profile counters across external scan files
(#69300)
### What problem does this PR solve?
Issue Number: None
Problem Summary: External small-file scans repeatedly initialize the
same Parquet profile counter hierarchy for each file, contending on the
shared RuntimeProfile registration lock. FileReader::init also repeats
the concrete reader's initialization. Initialize counters once per scan
instance and pass the immutable handles through table-reader wrappers to
Parquet readers. Keep distinct scan instances separate. Group the shared
Variant counters into one retained bundle so deferred materialization
can safely outlive the scanner without copying a shared_ptr for every
counter.
This change applies to native Parquet reads across Hive, Iceberg, Hudi,
Paimon and Fluss table-reader wrappers. It does not add speculative I/O
or change the file reader's selection and decoding behavior.
### Main benefit
**Avoid rebuilding the same Parquet counters for every small file.** In
matched Release-BE experiments with the FE held fixed, warm complete-SQL
full scans improve by **73–85%**: Hive with 10,000 partitions falls from
**738.04 to 120.15 ms**, and Iceberg with 10,000 files from **503.25 to
81.46 ms**. The full table and independent-BE-only results are below.
The mechanism is scan-owned immutable counter handles, reused across
files and forwarded through the real table-reader factories. Nested
readers retain their own child-profile caches, preserving attribution.
Deferred Variant work retains one shared counter bundle rather than
copying each counter owner separately. File decoding, predicates and
delete application are unchanged.
The headline A/B table uses exactly the same FE source and settings in
both variants; only the Release BE package changes. The elapsed-time
reduction therefore measures this BE optimization. No standalone latency
claim is made for Hudi, Paimon or Fluss merely because their factory
paths forward the cache.
### Release note
Reduce profile initialization and counter ownership overhead when
scanning many small Parquet files.
### Check List (For Author)
Latest validation at `024ea81a947`: all 446 scoped
reader/profile/Variant ASAN tests passed, as did clang-format 16 and
clang-tidy 22. The Linux CI crash in the added Hudi factory test was
traced to bypassing `TableReader::init`; the fixture now follows normal
initialization instead of leaving scan parameters uninitialized. This
follow-up changes tests only. The Release performance measurements below
predate the later factory/child-profile review repairs and are not a
fresh A/B of this head. CI status remains tracked by the checks on this
PR.
- Test: Unit Test and Manual Test
- New profile-cache tests cover concurrent initialization, counter
aggregation, independent scan instances, no-profile readers and retained
Variant counter lifetime
- Reader tests cover one initialization per file and sharing across
files
- run-be-ut.sh: 585 tests across 13 Parquet, ORC, Variant and
Hive/Iceberg/Hudi/Paimon/table-reader suites passed with ASAN
- Fresh Linux AArch64 build.sh --be -j3 baseline and optimized RELEASE
builds passed
- clang-format16 and header hygiene passed; preset clang-tidy passed all
19 changed files against the matching ASAN compilation database
- Isolated Release SQL results and a comparison with identical FE source
are reported below
- Behavior changed: No; preserve results and profile counter semantics
- Does this need documentation: No
### Performance validation
Complete SQL medians in milliseconds, matching Linux AArch64 Release BE,
single backend, 6 CPUs, warm footer/data cache. Both settings are fixed:
`split_batch_size=1`, `split_queue_capacity=4096`. Each single-query
result has 40 measured samples in forward/reverse order after identical
warmup. Cold denotes catalog/manifest cache invalidation, not a cold JVM
or BE data cache. COUNT and SUM results are checked, and post-timing
profiles verify footer/page hits.
To isolate the BE change from a pre-existing FE EOF polling race, both
columns below use exactly the same FE source containing its notification
fix; only the Release BE package changes.
| Warm full-scan scenario | Original BE, same fixed FE | This BE, same
fixed FE | Reduction |
|---|---:|---:|---:|
| Hive 1,000 partitions | 60.77 | 16.39 | 73.0% |
| Hive 10,000 partitions | 738.04 | 120.15 | 83.7% |
| Hive one partition / 10,000 files | 495.15 | 74.42 | 85.0% |
| Iceberg 1,000 manifests / 10,000 files | 503.25 | 81.46 | 83.8% |
A separate BE-only experiment on an older, unchanged FE measured
single-query medians of 22.89 ms for Hive 1,000 partitions, 230.37 ms
for Hive 10,000 partitions, 117.95 ms for Hive one partition with 10,000
files, and 138.38 ms for Iceberg 1,000 manifests. That FE had an
existing terminal-queue polling delay: a 400-SQL diagnostic found 78
delayed empty requests, including one whose production ended at 0.148 ms
but whose response returned at 100.911 ms. The headline table above uses
a fixed FE without that delay in both columns, so the BE comparison is
not dominated by FE polling. The two experiments use different fixed FE
versions and their absolute timings should not be mixed.
---
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 | 67 ++++----
be/src/format_v2/parquet/parquet_profile.h | 73 ++++++---
be/src/format_v2/parquet/parquet_reader.cpp | 10 +-
be/src/format_v2/parquet/parquet_reader.h | 3 +-
.../parquet/reader/variant_column_reader.cpp | 60 +++++---
be/src/format_v2/table/fluss_hybrid_reader.cpp | 1 +
be/src/format_v2/table/fluss_union_lake_reader.cpp | 2 +
be/src/format_v2/table/hudi_reader.cpp | 3 +-
.../iceberg_position_delete_sys_table_reader.cpp | 11 ++
.../iceberg_position_delete_sys_table_reader.h | 2 +
be/src/format_v2/table/iceberg_reader.cpp | 3 +-
be/src/format_v2/table/paimon_reader.cpp | 1 +
be/src/format_v2/table_reader.cpp | 3 +-
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 | 49 ++++++
.../parquet/variant_column_reader_test.cpp | 8 +-
be/test/format_v2/table/hudi_reader_test.cpp | 31 ++++
...eberg_position_delete_sys_table_reader_test.cpp | 45 ++++++
be/test/format_v2/table/iceberg_reader_test.cpp | 37 +++++
25 files changed, 517 insertions(+), 93 deletions(-)
diff --git a/be/src/exec/operator/file_scan_operator.cpp
b/be/src/exec/operator/file_scan_operator.cpp
index c72e8bfab3b..de0cc940ad5 100644
--- a/be/src/exec/operator/file_scan_operator.cpp
+++ b/be/src/exec/operator/file_scan_operator.cpp
@@ -35,6 +35,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"
@@ -69,6 +70,11 @@ int file_scanners_per_instance(RuntimeState* state) {
} // 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 {
@@ -232,6 +238,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 5250ee35ec3..e8aa83db088 100644
--- a/be/src/exec/operator/file_scan_operator.h
+++ b/be/src/exec/operator/file_scan_operator.h
@@ -27,6 +27,10 @@
#include "exec/scan/split_source_connector.h"
#include "format/format_common.h"
+namespace doris::format::parquet {
+class ParquetProfileCache;
+}
+
namespace doris {
class FileScanner;
class FileScannerV2;
@@ -40,8 +44,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;
@@ -85,6 +89,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;
TupleId _output_tuple_id = -1;
};
diff --git a/be/src/exec/scan/file_scanner_v2.cpp
b/be/src/exec/scan/file_scanner_v2.cpp
index fe14f8d69e3..c20e2f92cb5 100644
--- a/be/src/exec/scan/file_scanner_v2.cpp
+++ b/be/src/exec/scan/file_scanner_v2.cpp
@@ -400,7 +400,8 @@ FileScannerV2::FileScannerV2(RuntimeState* state,
FileScanLocalState* local_stat
const std::unordered_map<std::string, int>*
colname_to_slot_id)
: Scanner(state, local_state, limit, profile),
_split_source(std::move(split_source)),
- _kv_cache(kv_cache) {
+ _kv_cache(kv_cache),
+ _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) {
@@ -660,6 +661,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 25e2a433d13..0c029838d67 100644
--- a/be/src/exec/scan/file_scanner_v2.h
+++ b/be/src/exec/scan/file_scanner_v2.h
@@ -222,6 +222,7 @@ private:
std::unique_ptr<io::FileReaderStats> _file_reader_stats;
std::shared_ptr<io::IOContext> _io_ctx;
ShardedKVCache* _kv_cache = nullptr;
+ 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 2d94a23d673..95c272a4918 100644
--- a/be/src/format_v2/file_reader.cpp
+++ b/be/src/format_v2/file_reader.cpp
@@ -97,7 +97,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 6430ce894d3..fd67b5f70e2 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,
@@ -107,34 +121,36 @@ 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_unsupported_fallbacks = add_persistent_counter(
+ 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",
@@ -363,20 +379,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_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 22f8d7ad425..a6be04fda19 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
@@ -57,6 +54,24 @@ struct ParquetColumnReaderProfile {
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_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;
@@ -190,21 +205,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_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;
@@ -288,4 +296,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 6ab5f1bd037..bf528e6e693 100644
--- a/be/src/format_v2/parquet/parquet_reader.cpp
+++ b/be/src/format_v2/parquet/parquet_reader.cpp
@@ -575,8 +575,9 @@
ParquetReader::ParquetReader(std::shared_ptr<io::FileSystemProperties>& system_p
std::optional<format::GlobalRowIdContext>
global_rowid_context,
bool enable_mapping_timestamp_tz, bool
enable_mapping_varbinary,
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),
@@ -1098,7 +1099,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 64801aff7bf..b6cd2adaad0 100644
--- a/be/src/format_v2/parquet/parquet_reader.h
+++ b/be/src/format_v2/parquet/parquet_reader.h
@@ -49,7 +49,7 @@ public:
std::optional<format::GlobalRowIdContext>
global_rowid_context = std::nullopt,
bool enable_mapping_timestamp_tz = false, bool
enable_mapping_varbinary = false,
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;
@@ -93,6 +93,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 3c84d03ead4..c8729aaaac1 100644
--- a/be/src/format_v2/parquet/reader/variant_column_reader.cpp
+++ b/be/src/format_v2/parquet/reader/variant_column_reader.cpp
@@ -566,7 +566,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.
@@ -578,13 +579,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 {
@@ -1417,7 +1420,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()) {
@@ -1451,7 +1454,8 @@ public:
static_cast<bool>(residual) && has_present_value(residual)) {
// A residual value can contribute data to the same logical
object. Reconstructing
// is required in that case; returning only the typed leaf
would drop information.
-
update_counter(_profile.variant_direct_leaf_residual_fallbacks, 1);
+
update_counter(_profile.variant_profile().variant_direct_leaf_residual_fallbacks,
+ 1);
return std::nullopt;
}
@@ -1462,7 +1466,9 @@ public:
if (position + 1 == path.size()) {
if (typed_schema->kind != ParquetColumnSchemaKind::PRIMITIVE ||
check_and_get_column<ColumnNullable>(*typed) == nullptr) {
-
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.
@@ -1475,19 +1481,21 @@ public:
}
if (!supports_direct_typed_variant_state(*typed_schema)) {
if (_complete) {
-
update_counter(_profile.variant_direct_leaf_unsupported_fallbacks, 1);
+ update_counter(_profile.variant_profile()
+
.variant_direct_leaf_unsupported_fallbacks,
+ 1);
return std::nullopt;
}
// A partial projection cannot reconstruct its root.
Normalize only the exact
// requested leaf so Parquet annotations survive
heterogeneous file schemas.
- update_counter(_profile.variant_direct_leaf_rows,
+
update_counter(_profile.variant_profile().variant_direct_leaf_rows,
static_cast<int64_t>(typed->size()));
return VariantShreddedTypedValue {
.column = nullptr,
.type = nullptr,
.normalized =
normalize_projected_primitive_leaf(*typed_schema, typed)};
}
- update_counter(_profile.variant_direct_leaf_rows,
+
update_counter(_profile.variant_profile().variant_direct_leaf_rows,
static_cast<int64_t>(typed->size()));
return VariantShreddedTypedValue {.column = std::move(typed),
.type =
remove_nullable(typed_schema->type),
@@ -1570,21 +1578,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;
@@ -1670,7 +1681,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,
@@ -1681,22 +1692,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/fluss_hybrid_reader.cpp
b/be/src/format_v2/table/fluss_hybrid_reader.cpp
index 7c52b783543..5bd6a9a0620 100644
--- a/be/src/format_v2/table/fluss_hybrid_reader.cpp
+++ b/be/src/format_v2/table/fluss_hybrid_reader.cpp
@@ -277,6 +277,7 @@ Status
FlussHybridReader::_init_child_reader(format::TableReader* reader,
.io_ctx = _io_ctx,
.runtime_state = _runtime_state,
.scanner_profile = _scanner_profile,
+ .parquet_profile_cache = _parquet_profile_cache,
.file_slot_descs = _file_slot_descs,
.push_down_agg_type = _push_down_agg_type,
.push_down_count_columns = _push_down_count_columns,
diff --git a/be/src/format_v2/table/fluss_union_lake_reader.cpp
b/be/src/format_v2/table/fluss_union_lake_reader.cpp
index be65f5493b7..d49803943f3 100644
--- a/be/src/format_v2/table/fluss_union_lake_reader.cpp
+++ b/be/src/format_v2/table/fluss_union_lake_reader.cpp
@@ -175,6 +175,7 @@ Status
FlussUnionLakeReader::init(format::TableReadOptions&& options) {
.io_ctx = _io_ctx,
.runtime_state = _runtime_state,
.scanner_profile = _scanner_profile,
+ .parquet_profile_cache = _parquet_profile_cache,
.file_slot_descs = _file_slot_descs,
// Aggregate pushdown is withheld from the lake half on purpose. A
COUNT answered from
// paimon's own file metadata would count the very rows this
reader is about to suppress,
@@ -468,6 +469,7 @@ Status FlussUnionLakeReader::_read_tail_keys(const Tail&
tail, SuppressionKeys*
.io_ctx = _io_ctx,
.runtime_state = _runtime_state,
.scanner_profile = _scanner_profile,
+ .parquet_profile_cache = _parquet_profile_cache,
}));
// Not the adaptive size the lake half is being read at: this is a bounded
pass over key columns
// only, and its batches have nothing to do with how wide the lake's rows
turned out to be.
diff --git a/be/src/format_v2/table/hudi_reader.cpp
b/be/src/format_v2/table/hudi_reader.cpp
index 0edb76a007e..caef984c5b5 100644
--- a/be/src/format_v2/table/hudi_reader.cpp
+++ b/be/src/format_v2/table/hudi_reader.cpp
@@ -75,7 +75,7 @@ Status
HudiReader::create_file_reader(std::unique_ptr<format::FileReader>* reade
*reader = std::make_unique<format::parquet::ParquetReader>(
_system_properties, _current_task->data_file, _io_ctx,
_scanner_profile,
_global_rowid_context, enable_mapping_timestamp_tz,
enable_mapping_varbinary,
- parquet_int96_time_zone());
+ parquet_int96_time_zone(), false, _parquet_profile_cache);
return Status::OK();
}
@@ -219,6 +219,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 e6ef149e2d5..c84cf23743f 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 1068567076b..b1e1e704457 100644
--- a/be/src/format_v2/table/iceberg_reader.cpp
+++ b/be/src/format_v2/table/iceberg_reader.cpp
@@ -1666,7 +1666,8 @@ Status
IcebergTableReader::_create_delete_file_reader(const TIcebergDeleteFileDe
// Delete and data files must parse raw binary fields with the same
scan-level mapping.
*reader = std::make_unique<format::parquet::ParquetReader>(
system_properties, file_description, io_ctx, _scanner_profile,
std::nullopt,
- enable_mapping_timestamp_tz, enable_mapping_varbinary, "",
preserve_binary_uuid());
+ enable_mapping_timestamp_tz, enable_mapping_varbinary, "",
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/paimon_reader.cpp
b/be/src/format_v2/table/paimon_reader.cpp
index 794d2443357..4473240011d 100644
--- a/be/src/format_v2/table/paimon_reader.cpp
+++ b/be/src/format_v2/table/paimon_reader.cpp
@@ -596,6 +596,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 a51140143ba..5a8d5dea5e7 100644
--- a/be/src/format_v2/table_reader.cpp
+++ b/be/src/format_v2/table_reader.cpp
@@ -959,6 +959,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;
@@ -1409,7 +1410,7 @@ Status
TableReader::create_file_reader(std::unique_ptr<FileReader>* reader) {
*reader = std::make_unique<format::parquet::ParquetReader>(
_system_properties, _current_task->data_file, _io_ctx,
_scanner_profile,
_global_rowid_context, enable_mapping_timestamp_tz,
enable_mapping_varbinary,
- hive_parquet_time_zone, preserve_binary_uuid());
+ 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 928254b177e..11fa915c77a 100644
--- a/be/src/format_v2/table_reader.h
+++ b/be/src/format_v2/table_reader.h
@@ -75,6 +75,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
@@ -148,6 +152,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;
@@ -2236,6 +2241,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 1ff550bce69..027f20ae89f 100644
--- a/be/test/format_v2/parquet/parquet_reader_test.cpp
+++ b/be/test/format_v2/parquet/parquet_reader_test.cpp
@@ -1973,6 +1973,55 @@ 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, 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 a36d1cdddfd..43772eff06b 100644
--- a/be/test/format_v2/parquet/variant_column_reader_test.cpp
+++ b/be/test/format_v2/parquet/variant_column_reader_test.cpp
@@ -2615,10 +2615,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 1a4b51a8249..40247dab31a 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,6 +157,36 @@ public:
// 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);
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 6db534fa06d..d6e3efce7f4 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
@@ -21,6 +21,8 @@
#include "core/data_type/data_type_number.h"
#include "core/data_type/data_type_struct.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"
@@ -61,6 +63,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;
const auto child_type = make_nullable(std::make_shared<DataTypeInt32>());
diff --git a/be/test/format_v2/table/iceberg_reader_test.cpp
b/be/test/format_v2/table/iceberg_reader_test.cpp
index 9b70c0ada25..9a7af2a574d 100644
--- a/be/test/format_v2/table/iceberg_reader_test.cpp
+++ b/be/test/format_v2/table/iceberg_reader_test.cpp
@@ -74,6 +74,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"
@@ -6503,5 +6504,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
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
