This is an automated email from the ASF dual-hosted git repository.
morningman 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 1cd984aa5f3 [opt](build) Cut three more waves of hot include edges in
the BE header graph (#66672)
1cd984aa5f3 is described below
commit 1cd984aa5f3e54417aad9832e59e41f63dcc8b8d
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Thu Aug 13 11:54:55 2026 +0800
[opt](build) Cut three more waves of hot include edges in the BE header
graph (#66672)
> Split out of **https://github.com/apache/doris/pull/66510**, which
carries the whole
> BE build-time batch. This PR continues the header-closure surgery
merged as
> **#66400** with three more waves, and is independent of the rest of
that batch —
> it can be reviewed and merged on its own.
### What problem does this PR solve?
Related PR: #66400, #66510
Problem Summary:
Three waves of include-edge surgery in the BE header graph, each in the
same
"seed first, then cut" shape as #66400: one purely additive commit that
outlines
the inline bodies forcing the instantiation and seeds the direct
includes the cut
will expose, then one commit that deletes the edges and pins them with
new
`build-support/check-header-deps.py` rules.
#### Wave 1 — three hot edges (`Prepare cutting three hot edges` + `Cut
three hot include edges`)
| edge cut | why it cost so much | how it is made removable |
|---|---|---|
| `core/pod_array.h` → `runtime/thread_context.h` | **dead include**
left over from the PODArray memory-tracking experiment (#50549); the
tracking logic has since moved into `Allocator` and `pod_array.h` names
no `thread_context` symbol | just delete it — 203 TUs stop seeing
`thread_context.h` and 195 of them stop seeing `exec_env.h` (1.37 MB/TU
differential payload) |
| `core/column/column.h` → `exec/sort/hybrid_sorter.h` | a `core` →
`exec` layering violation reaching **808 TUs** | `HybridSorter` only
appears in virtual signatures, so a forward declaration covers it; the
`BE_TEST`-only `get_permutation_default` body (it constructs a
`HybridSorter` by value) moves to `column.cpp` |
| `core/wide_integer_impl.h` → `boost/multiprecision` (unconditional) |
4.27 MB of preprocessed closure in **1169 TUs** on every platform
without an 80-bit long double | the from-double members stay inline (and
`constexpr`) only where `LDBL_MANT_DIG == 64`; elsewhere they compile
once in a new `core/wide_integer_from_double.cpp` with explicit
instantiations for `integer<128\|256, signed\|unsigned>`. They were
never `constexpr` on those platforms, so no constant evaluation is lost
|
Preprocessed closure of `column.h`: **16.72 MB → 10.15 MB (-39%)**.
#### Wave 2 — two instantiation amplifiers: RLE and FMT_COMPILE
Both parquet `decoder.h` trees included `util/rle_encoding.h` only
because inline
`BaseDictDecoder` bodies made ~530 TUs instantiate the
`RleBatchDecoder<uint32_t>` → `GetLiteralValues` → `UnpackBatch` →
`UnpackValues`
chain — measured at **0.43 CPU s per TU, 231 CPU s total** — on top of
reparsing
the 2038-line rle/bit-stream/bit-packing family each time. The eight
per-page /
per-batch methods plus the ctor and dtor move out of line, so the
`unique_ptr<RleBatchDecoder<uint32_t>>` member only needs the complete
type in
`decoder.cpp`. Everything moved is dispatched through the vtable at
every call
site already, so no generated call changes; the real decode TUs keep
including
`rle_encoding.h` directly and inline the chain exactly as before, and
the
per-value-hot `LevelDecoder::get_next` path is deliberately untouched.
`core/uint24.h` and `core/value/large_int_value.h` pushed their
`FMT_COMPILE`
formatter instantiations (the `"{:04d}-{:02d}-{:02d}"` date formatter is
the
single biggest fmt instantiation in the tree, plus the int128
formatters) into
~1150 TUs, **53.5 CPU s**. Those bodies move to `.cpp` files and both
headers
drop `<fmt/compile.h>` / `<fmt/format.h>`. They return `std::string` and
are
allocation-dominated, so the now-outlined call is noise.
#### Wave 3 — the DataVariants amplifier behind `dependency.h`
`exec/pipeline/dependency.h` and `exec/pipeline/rec_cte_shared_state.h`
hold
non-template **in-class inline** bodies — SharedState constructors,
destructors,
close paths and three `std::visit` dispatches — that name the full
`AggregatedDataVariants` / `JoinDataVariants` / `SetDataVariants` /
`DistinctDataVariants` surface. Such bodies are semantically analyzed
**when the
header is parsed**, not when they are called, so every one of the ~128
TUs that
transitively include `dependency.h` instantiated that whole surface at a
flat
**~0.85 CPU s per TU (~130 CPU s)** — and 105 of those TUs never touch a
variant.
All of it is per-query setup/teardown, so it moves to `dependency.cpp`
and a new
`rec_cte_shared_state.cpp`. With the bodies gone the headers can drop:
- `exec/common/agg_utils.h`, `set_utils.h`, `distinct_agg_utils.h` —
forward declarations suffice;
- `exec/common/join_utils.h` → the new light
`exec/common/join_op_utils.h` (JoinOpVariants and the `AsofIndexGroup`
family split out of `join_utils.h`, which re-exports them;
`dependency.h` holds these **by value**, and the new header depends only
on thrift enums, pdqsort and std containers — not on the hash tables);
- `exec/operator/join/process_hash_table_probe.h` — dead include;
- `util/brpc_closure.h` — dead include, and the sole route carrying
`query_context.h`, `thread_context.h` and `service/brpc.h` into ~100
pipeline TUs (1.36 MB/TU);
- `<concurrentqueue.h>` — dead include (152 KB third-party single
header);
- `util/brpc_client_cache.h` from `rec_cte_shared_state.h` — the rpc
send bodies live in the `.cpp` now.
`BucketedAggSharedState::init_instances` also becomes a non-template
taking
`std::function`: non-dependent constructs in a member-template body are
checked
at definition time, so the old inline template forced the destructor of
`unique_ptr<BucketedAggDataVariants>` on every includer despite never
being
called there.
#### One CI-critical commit rides along
`Opt wide_integer_from_double.cpp out of the PCH` must ship **in this
PR**, not as
a follow-up. Upstream's clang toolchain defaults to `ENABLE_PCH=ON`
(`be/CMakeLists.txt`), and `pch.h` transitively includes
`wide_integer_impl.h`,
whose include guard is then already consumed when the impl TU is
compiled — the
explicit instantiations would find no definition. It is the only file in
the tree
with this interaction, but splitting the two commits apart would leave a
state
that fails the clang CI lane.
### Benefit
Compile-time only; no runtime behavior change.
#### Head-to-head on exactly this PR's content
Cold, cache-free BE builds of this PR's merge base (`c29075a7e10`) and
its head
(`03129023ee5`), run back-to-back on the same machine with the repo's
own
`--compile-bench` harness — dedicated always-cold build dirs, ccache
disabled,
`-j10`, `ENABLE_PCH=ON`, and both runs started from the same cooled
state
(load1 2.1 vs 2.3) so neither side pays for the other's heat.
| metric | before | after | delta |
|---|---|---|---|
| **build phase wall** | 14m50s | **14m14s** | **-36.3 s (-4.1%)** |
| **Σ TU wall time** (parallelism-independent) | 2h19m | **2h13m** |
**-6 min (-4.3%)** |
| Σ TU cpu (user+sys) | 2h18m | 2h12m | -6 min |
| effective parallelism | 9.4× | 9.4× | identical ⇒ clean attribution |
| per-file wall | — | — | **197 improved / 12 regressed / 3 new** |
The improvements land exactly on the predicted targets —
`pipeline_fragment_context.cpp`
-6.0 s, `rec_cte_anchor_sink_operator.cpp` -5.0 s, `operator.cpp` -3.7
s,
`data_queue.cpp` -3.0 s (all W3 `dependency.h` consumers), and
`byte_array_dict_decoder.cpp` -2.5 s (the W2 decoder edge).
Of the 12 regressions, **`dependency.cpp` +2.6 s is by design**: that is
the TU the
SharedState bodies were moved *into*, so it pays once for what ~128 TUs
stop paying.
The rest sit in the noise floor of a `-j10` run, and there is direct
evidence for
that floor: `gensrc/build/gen_cpp/cloud.pb.cc`, a generated protobuf TU
this PR
cannot touch, moved **+2.3 s** between the two runs.
**Why no end-to-end total-wall number is quoted:** the two trees
differed *outside*
the build phase — the baseline tree skipped the contrib submodule step
while the PR
tree re-fetched it (+58.1 s), and its gensrc was already generated (-5.6
s). Those
phases are not compilation, so only the build-phase and Σ-TU figures
above are
attributable to the change.
**One caveat worth stating:** these numbers are with `ENABLE_PCH=ON`. A
PCH already
amortizes exactly the kind of shared headers this PR is cutting, so it
masks part of
the win — and the upstream compile lane builds *without* a PCH (its
command line
carries no `-include-pch`). The effect there should be larger, not
smaller.
#### Per-wave numbers from the development branch
| wave | effect | measurement |
|---|---|---|
| W1 | **-24.3 s** wall, `be/src` CPU **-3.6%** | same-codebase A/B on
the batch branch |
| W2 | **-125 s / -6.9%** (30m24s → 28m19s) | same-codebase A/B,
back-to-back |
| W3 | ~130 CPU s of parse-time instantiation removed (~0.85 CPU s ×
~128 TUs), plus 1.36 MB/TU × ~100 TUs of dead brpc payload | P3.5
research traces (per-TU CPU, not end-to-end wall) |
These were taken on the batch branch at `-j5`/`-j6` against an older
base, so they
do not add up to the head-to-head figure above; they are included
because they
attribute the win to each wave separately.
### Risk and verification
- **Header sweeps.** Every wave was validated with a full
`-fsyntax-only` sweep over the natural (no-PCH) include closure: W1 1358
TUs with only the 4 known pre-existing failures, W2 **1364/1364 clean**,
W3 **0 failing of 1365**.
- **BE unit tests.** The first sweeps covered `be/src` only, so a full
`ninja -k 0` over 2422 targets was run to reach the 1024 `be/test` TUs;
it surfaced 8 failing TUs in 4 families, all repaired in `Repair the BE
UT build after the include-edge cuts`. **Nothing there restores a cut
edge and no production header gains an include.** Two of those repairs
fix a latent defect that predates this PR: `BaseDictDecoder`'s
defaulted-in-class constructor odr-uses
`~unique_ptr<RleBatchDecoder<uint32_t>>` in every TU constructing a
derived decoder, so the header's own claim that the complete type is
only needed in `decoder.cpp` did not hold — in **both** trees, though
only the `format/` one had a test reaching it.
- **Rebased onto current master (`c29075a7e10`) and rebuilt from
scratch** in a clean worktree, macOS/arm64 + clang 20, `ENABLE_PCH=ON`
(upstream's clang default, so the PCH opt-out above is exercised):
**8554/8554 ninja edges, zero failures, `doris_be` links.**
`compile_commands.json` confirms `wide_integer_from_double.cpp` is the
one first-party TU compiled without the PCH.
- **BE unit tests rebuilt on the rebased tree: all 9602 objects compile,
zero failures**, and the final link resolves every symbol (0 undefined).
Two pre-existing macOS-only obstacles were hit on the way and are called
out under *Proactive disclosure*.
- **Natural-closure sweep (`syntax_sweep.py --no-pch`, the gate added in
#66616): 0 failing TUs of 1380.** This is the gate that matters for a PR
that cuts include edges — a normal build with `ENABLE_PCH=ON` cannot see
a missing include that the PCH happens to supply, which is precisely how
the `<ranges>` breakage in the last commit reached CI before it reached
me.
- **Runtime cost of outlining.** Everything moved out of line is either
per-query setup/teardown (SharedState ctor/dtor/close), or a
per-page/per-batch method already reached through a vtable. No per-value
or per-row hot path was outlined.
- **Guard rules.** `check-header-deps.py` gains rules pinning
`pod_array.h !-> thread_context.h`, `column.h !-> exec/sort/`, and both
`decoder.h` headers `!-> util/rle_encoding.h` (19 rules total, all
passing).
### Proactive disclosure
- **One commit here is unrelated to include edges: `Make
hierarchical_data_iterator_test compile on macOS arm64`.**
`std::min(*rows, ROWS - current_ordinal)` cannot deduce `_Tp` where
`size_t` is `unsigned long` and `ordinal_t` (`uint64_t`) is `unsigned
long long`, which is the case on macOS/arm64 but not on Linux — so CI is
green while `be/test` does not build on macOS at all. It arrived with
#66204. It is fixed here because this branch is verified on macOS and
the break blocks that verification; happy to split it out if a reviewer
prefers.
- **The macOS Debug UT link has outgrown the Mach-O format**,
independently of this PR: `section __debug_names's file offset exceeds
4GB`. Master's own test growth is what crossed the line — the same
worktree layout linked fine on 2026-08-10 at 7.6 GB of test debug info,
and master is now at 8.0 GB across 10 more test TUs, while this PR adds
one `#include` to each of 5 test files. Omitting the debug map
(`-Wl,-S`) links the binary cleanly with zero undefined symbols, which
is how the link was verified here. Worth someone's attention as a
separate issue.
- **Cross-platform is the blind spot.** All measurements and all sweeps
ran on macOS/arm64 + clang 20. Nothing here is platform-specific by
construction, but the Linux gcc/clang lanes are covered only by upstream
CI, not by any local gate — please give those two lanes a look.
- **Two cuts have no scanner rule.** The boost/multiprecision edge is
preprocessor-gated and `check-header-deps.py` is preprocessor-blind (it
would flag the impl TU's gated include); the macro structure is
self-guarding instead — breaking it fails the impl TU's build. The fmt
cuts are likewise only noted in a comment, since the scanner follows
quoted project includes.
- **Three test files carry `clang-format off/on` around their
includes.** `asof_join_test` and the two `fix_length_dict_decoder` tests
are include-order-sensitive: the supplying include has to come *before*
the header under test (ADL cannot reach the global `pdqsort` from
`std::vector`'s iterators; a `unique_ptr<RleBatchDecoder<uint32_t>>`
dereference depends on no template parameter, so it binds where the
template is parsed). Without the marker the formatter sorts the include
back and breaks the build. The cost was deliberately kept in the tests
rather than paid by a production header.
- **`DISALLOW_COPY_AND_ASSIGN` in `storage/olap_define.h` loses its
trailing semicolon.** `butil/macros.h` defines the same macro without
one and wins under `#ifndef` in TUs that see butil first, so the two
expansions have to stay call-site compatible. All 45 call sites in the
tree already write the `;`.
---
be/src/cloud/cloud_meta_mgr.cpp | 1 +
be/src/cloud/cloud_meta_mgr.h | 1 +
be/src/core/CMakeLists.txt | 7 +
be/src/core/column/column.cpp | 9 +
be/src/core/column/column.h | 9 +-
be/src/core/column/column_const.h | 1 +
be/src/core/column/column_decimal.cpp | 28 +++
be/src/core/column/column_decimal.h | 26 +--
be/src/core/pod_array.h | 1 -
.../core/{value/large_int_value.cpp => uint24.cpp} | 34 +---
be/src/core/uint24.h | 13 +-
be/src/core/value/large_int_value.cpp | 15 ++
be/src/core/value/large_int_value.h | 16 +-
be/src/core/wide_integer_from_double.cpp | 50 +++++
be/src/core/wide_integer_impl.h | 32 ++-
be/src/exec/common/join_op_utils.h | 157 ++++++++++++++
be/src/exec/common/join_utils.h | 122 +----------
be/src/exec/exchange/local_exchanger.h | 2 +
be/src/exec/operator/aggregation_sink_operator.h | 1 +
.../exec/operator/aggregation_source_operator.cpp | 1 +
.../operator/bucketed_aggregation_sink_operator.h | 1 +
.../bucketed_aggregation_source_operator.cpp | 1 +
be/src/exec/operator/hashjoin_build_sink.cpp | 1 +
be/src/exec/operator/hashjoin_probe_operator.cpp | 1 +
be/src/exec/operator/hashjoin_probe_operator.h | 1 +
.../exec/operator/join/process_hash_table_probe.h | 1 +
.../operator/join/process_hash_table_probe_impl.h | 1 +
be/src/exec/operator/join_build_sink_operator.h | 1 +
be/src/exec/operator/materialization_opertor.h | 1 +
.../partitioned_aggregation_source_operator.cpp | 1 +
be/src/exec/operator/rec_cte_source_operator.cpp | 2 +
be/src/exec/operator/set_probe_sink_operator.cpp | 1 +
be/src/exec/operator/set_sink_operator.cpp | 1 +
be/src/exec/operator/set_source_operator.cpp | 1 +
.../exec/operator/streaming_aggregation_operator.h | 1 +
be/src/exec/pipeline/dependency.cpp | 132 ++++++++++++
be/src/exec/pipeline/dependency.h | 174 +++++-----------
be/src/exec/pipeline/rec_cte_shared_state.cpp | 157 ++++++++++++++
be/src/exec/pipeline/rec_cte_shared_state.h | 141 ++-----------
be/src/format/parquet/decoder.cpp | 23 +++
be/src/format/parquet/decoder.h | 28 +--
be/src/format_v2/native/native_reader.h | 1 +
be/src/format_v2/parquet/reader/native/decoder.cpp | 219 ++++++++++++++++++++
be/src/format_v2/parquet/reader/native/decoder.h | 225 ++-------------------
be/src/storage/compaction/compaction.cpp | 1 +
be/src/storage/index/ann/ann_index_writer.h | 11 +
be/src/storage/index/index_file_writer.h | 11 +
.../index/inverted/inverted_index_searcher.h | 11 +
be/src/storage/index/inverted/query/query.h | 11 +
.../index/inverted/similarity/bm25_similarity.cpp | 2 -
be/src/storage/olap_define.h | 7 +-
be/src/storage/tablet/base_tablet.cpp | 1 +
be/test/core/block/block_test.cpp | 1 +
be/test/core/column/column_variant_v2_test.cpp | 1 +
be/test/exec/operator/asof_join_test.cpp | 7 +
.../fix_length_dict_decoder_empty_dict_test.cpp | 9 +
.../parquet/fix_length_dict_decoder_test.cpp | 9 +
.../segment/hierarchical_data_iterator_test.cpp | 6 +-
build-support/check-header-deps.py | 102 ++++++++++
59 files changed, 1153 insertions(+), 679 deletions(-)
diff --git a/be/src/cloud/cloud_meta_mgr.cpp b/be/src/cloud/cloud_meta_mgr.cpp
index 0a2ebe46635..7a8b2ee8ff2 100644
--- a/be/src/cloud/cloud_meta_mgr.cpp
+++ b/be/src/cloud/cloud_meta_mgr.cpp
@@ -37,6 +37,7 @@
#include <memory>
#include <mutex>
#include <random>
+#include <ranges>
#include <shared_mutex>
#include <string>
#include <type_traits>
diff --git a/be/src/cloud/cloud_meta_mgr.h b/be/src/cloud/cloud_meta_mgr.h
index 4b3f3bed044..1bcfbe69a81 100644
--- a/be/src/cloud/cloud_meta_mgr.h
+++ b/be/src/cloud/cloud_meta_mgr.h
@@ -20,6 +20,7 @@
#include <future>
#include <memory>
+#include <ranges>
#include <string>
#include <tuple>
#include <variant>
diff --git a/be/src/core/CMakeLists.txt b/be/src/core/CMakeLists.txt
index 84de7cd3a9d..348a375eef0 100644
--- a/be/src/core/CMakeLists.txt
+++ b/be/src/core/CMakeLists.txt
@@ -25,3 +25,10 @@ file(GLOB_RECURSE SRC_FILES CONFIGURE_DEPENDS *.cpp)
add_library(Core STATIC ${SRC_FILES})
pch_reuse(Core)
+# wide_integer_from_double.cpp defines DORIS_WIDE_INTEGER_FROM_DOUBLE_IMPL_TU
+# before including wide_integer_impl.h to compile the from-double bodies
exactly
+# once. The PCH already contains that header (via storage/olap_common.h) in
+# declaration-only mode, and its pre-spent include guard would leave the
explicit
+# instantiations below without definitions. Opt this single TU out of the PCH.
+set_source_files_properties(${CMAKE_CURRENT_SOURCE_DIR}/wide_integer_from_double.cpp
+ PROPERTIES SKIP_PRECOMPILE_HEADERS ON)
diff --git a/be/src/core/column/column.cpp b/be/src/core/column/column.cpp
index ce49646b839..46471fac2ac 100644
--- a/be/src/core/column/column.cpp
+++ b/be/src/core/column/column.cpp
@@ -23,6 +23,7 @@
#include "core/column/column_const.h"
#include "core/column/column_nullable.h"
#include "core/data_type/data_type.h"
+#include "exec/sort/hybrid_sorter.h"
#include "exec/sort/sort_block.h"
#include "util/simd/bits.h"
@@ -242,4 +243,12 @@ void IColumn::check_const_only_in_top_level() const {
for_each_subcolumn(throw_if_const);
}
+#ifdef BE_TEST
+void IColumn::get_permutation_default(bool reverse, size_t limit, int
nan_direction_hint,
+ Permutation& res) const {
+ HybridSorter sorter;
+ get_permutation(reverse, limit, nan_direction_hint, sorter, res);
+}
+#endif
+
} // namespace doris
diff --git a/be/src/core/column/column.h b/be/src/core/column/column.h
index 9a6f3d05d02..b3545309cbc 100644
--- a/be/src/core/column/column.h
+++ b/be/src/core/column/column.h
@@ -36,7 +36,6 @@
#include "core/string_ref.h"
#include "core/typeid_cast.h"
#include "core/types.h"
-#include "exec/sort/hybrid_sorter.h"
#include "storage/olap_common.h"
namespace doris {
@@ -47,6 +46,7 @@ namespace doris {
class Arena;
class ColumnSorter;
+class HybridSorter;
using EqualFlags = std::vector<uint8_t>;
using EqualRange = std::pair<int, int>;
@@ -514,11 +514,10 @@ public:
}
#ifdef BE_TEST
+ // Defined in column.cpp: constructs a HybridSorter, which is only
+ // forward-declared here to keep exec/sort out of column.h's closure.
void get_permutation_default(bool reverse, size_t limit, int
nan_direction_hint,
- Permutation& res) const {
- HybridSorter sorter;
- get_permutation(reverse, limit, nan_direction_hint, sorter, res);
- }
+ Permutation& res) const;
#endif
/** Split column to smaller columns. Each value goes to column index,
selected by corresponding element of 'selector'.
diff --git a/be/src/core/column/column_const.h
b/be/src/core/column/column_const.h
index 7c38fd2cbbe..44b1f9d477a 100644
--- a/be/src/core/column/column_const.h
+++ b/be/src/core/column/column_const.h
@@ -27,6 +27,7 @@
#include <cstdint>
#include <functional>
#include <initializer_list>
+#include <span>
#include <string>
#include <type_traits>
#include <utility>
diff --git a/be/src/core/column/column_decimal.cpp
b/be/src/core/column/column_decimal.cpp
index eed01445bf8..fed2fe04c3a 100644
--- a/be/src/core/column/column_decimal.cpp
+++ b/be/src/core/column/column_decimal.cpp
@@ -34,6 +34,7 @@
#include "core/value/decimalv2_value.h"
#include "exec/common/int_exp.h"
#include "exec/common/sip_hash.h"
+#include "exec/sort/hybrid_sorter.h"
#include "exec/sort/sort_block.h"
#include "util/hash_util.hpp"
#include "util/simd/bits.h"
@@ -305,6 +306,33 @@ Field ColumnDecimal<T>::operator[](size_t n) const {
return Field::create_field<T>(*(typename
PrimitiveTypeTraits<T>::CppType*)(&data[n]));
}
+template <PrimitiveType T>
+template <typename U>
+void ColumnDecimal<T>::permutation(bool reverse, size_t limit, HybridSorter&
sorter,
+ PaddedPODArray<U>& res) const {
+ size_t s = data.size();
+ res.resize(s);
+ for (U i = 0; i < s; ++i) res[i] = i;
+
+ auto sort_end = res.end();
+ if (limit && static_cast<double>(limit) < static_cast<double>(s) / 8.0) {
+ sort_end = res.begin() + limit;
+ if (reverse)
+ std::partial_sort(res.begin(), sort_end, res.end(),
+ [this](size_t a, size_t b) { return data[a] >
data[b]; });
+ else
+ std::partial_sort(res.begin(), sort_end, res.end(),
+ [this](size_t a, size_t b) { return data[a] <
data[b]; });
+ } else {
+ if (reverse)
+ sorter.sort(res.begin(), res.end(),
+ [this](size_t a, size_t b) { return data[a] > data[b];
});
+ else
+ sorter.sort(res.begin(), res.end(),
+ [this](size_t a, size_t b) { return data[a] < data[b];
});
+ }
+}
+
template <PrimitiveType T>
void ColumnDecimal<T>::get_permutation(bool reverse, size_t limit, int,
HybridSorter& sorter,
IColumn::Permutation& res) const {
diff --git a/be/src/core/column/column_decimal.h
b/be/src/core/column/column_decimal.h
index 0f2d96f7e34..9d6ec34c8b8 100644
--- a/be/src/core/column/column_decimal.h
+++ b/be/src/core/column/column_decimal.h
@@ -285,31 +285,11 @@ public:
protected:
Container data;
UInt32 scale;
+ // Defined in column_decimal.cpp (its only caller is get_permutation
there):
+ // the body dereferences HybridSorter, which is forward-declared in
column.h.
template <typename U>
void permutation(bool reverse, size_t limit, HybridSorter& sorter,
- PaddedPODArray<U>& res) const {
- size_t s = data.size();
- res.resize(s);
- for (U i = 0; i < s; ++i) res[i] = i;
-
- auto sort_end = res.end();
- if (limit && static_cast<double>(limit) < static_cast<double>(s) /
8.0) {
- sort_end = res.begin() + limit;
- if (reverse)
- std::partial_sort(res.begin(), sort_end, res.end(),
- [this](size_t a, size_t b) { return data[a]
> data[b]; });
- else
- std::partial_sort(res.begin(), sort_end, res.end(),
- [this](size_t a, size_t b) { return data[a]
< data[b]; });
- } else {
- if (reverse)
- sorter.sort(res.begin(), res.end(),
- [this](size_t a, size_t b) { return data[a] >
data[b]; });
- else
- sorter.sort(res.begin(), res.end(),
- [this](size_t a, size_t b) { return data[a] <
data[b]; });
- }
- }
+ PaddedPODArray<U>& res) const;
void ALWAYS_INLINE decimalv2_do_crc(size_t i, uint32_t& hash) const {
const auto& dec_val = (const DecimalV2Value&)data[i];
diff --git a/be/src/core/pod_array.h b/be/src/core/pod_array.h
index 2ca12dc49ed..4ce168b4858 100644
--- a/be/src/core/pod_array.h
+++ b/be/src/core/pod_array.h
@@ -35,7 +35,6 @@
#include "common/compiler_util.h" // IWYU pragma: keep
#include "core/allocator.h" // IWYU pragma: keep
#include "core/memcpy_small.h"
-#include "runtime/thread_context.h"
#ifndef NDEBUG
#include <sys/mman.h>
diff --git a/be/src/core/value/large_int_value.cpp b/be/src/core/uint24.cpp
similarity index 50%
copy from be/src/core/value/large_int_value.cpp
copy to be/src/core/uint24.cpp
index 4fac5285265..7688a02edcf 100644
--- a/be/src/core/value/large_int_value.cpp
+++ b/be/src/core/uint24.cpp
@@ -15,36 +15,20 @@
// specific language governing permissions and limitations
// under the License.
-#include "core/value/large_int_value.h"
+#include "core/uint24.h"
-#include <string>
-
-#include "util/string_parser.hpp"
+#include <fmt/compile.h>
+#include <fmt/format.h>
namespace doris {
-std::ostream& operator<<(std::ostream& os, __int128 const& value) {
- std::ostream::sentry s(os);
- if (s) {
- std::string value_str = fmt::format("{}", value);
- if (os.rdbuf()->sputn(value_str.data(), value_str.size()) !=
value_str.size()) {
- os.setstate(std::ios_base::badbit);
- }
- }
- return os;
-}
+std::string uint24_t::to_string() const {
+ int value = *reinterpret_cast<const uint24_t*>(data);
+ int mday = value & 31;
+ int mon = value >> 5 & 15;
+ int year = value >> 9;
-std::istream& operator>>(std::istream& is, __int128& value) {
- std::string str;
- is >> str;
- StringParser::ParseResult result;
- value = StringParser::string_to_int<__int128>(str.c_str(), str.size(),
&result);
- if (result != StringParser::PARSE_SUCCESS) {
- is.setstate(std::ios_base::failbit);
- }
- return is;
+ return fmt::format(FMT_COMPILE("{:04d}-{:02d}-{:02d}"), year, mon, mday);
}
} // namespace doris
-
-/* vim: set expandtab ts=4 sw=4 sts=4 tw=100: */
diff --git a/be/src/core/uint24.h b/be/src/core/uint24.h
index f2fc77ef7ab..a98144591ae 100644
--- a/be/src/core/uint24.h
+++ b/be/src/core/uint24.h
@@ -17,8 +17,6 @@
#pragma once
-#include <fmt/compile.h>
-
#include <cstdint>
#include <cstring>
#include <iostream>
@@ -128,14 +126,9 @@ public:
return 0;
}
- std::string to_string() const {
- int value = *reinterpret_cast<const uint24_t*>(data);
- int mday = value & 31;
- int mon = value >> 5 & 15;
- int year = value >> 9;
-
- return fmt::format(FMT_COMPILE("{:04d}-{:02d}-{:02d}"), year, mon,
mday);
- }
+ // Defined in uint24.cpp: the FMT_COMPILE format machinery is expensive to
+ // instantiate and this header is included nearly everywhere.
+ std::string to_string() const;
const uint8_t* get_data() const { return data; }
diff --git a/be/src/core/value/large_int_value.cpp
b/be/src/core/value/large_int_value.cpp
index 4fac5285265..37f64df2f02 100644
--- a/be/src/core/value/large_int_value.cpp
+++ b/be/src/core/value/large_int_value.cpp
@@ -17,12 +17,27 @@
#include "core/value/large_int_value.h"
+#include <fmt/compile.h>
+#include <fmt/format.h>
+
#include <string>
#include "util/string_parser.hpp"
namespace doris {
+int64_t LargeIntValue::to_buffer(__int128 value, char* buffer) {
+ return fmt::format_to(buffer, FMT_COMPILE("{}"), value) - buffer;
+}
+
+std::string LargeIntValue::to_string(__int128 value) {
+ return fmt::format(FMT_COMPILE("{}"), value);
+}
+
+std::string LargeIntValue::to_string(__uint128_t value) {
+ return fmt::format(FMT_COMPILE("{}"), value);
+}
+
std::ostream& operator<<(std::ostream& os, __int128 const& value) {
std::ostream::sentry s(os);
if (s) {
diff --git a/be/src/core/value/large_int_value.h
b/be/src/core/value/large_int_value.h
index c65879eb95a..9a2a4878401 100644
--- a/be/src/core/value/large_int_value.h
+++ b/be/src/core/value/large_int_value.h
@@ -16,8 +16,6 @@
// under the License.
#pragma once
-#include <fmt/compile.h>
-#include <fmt/format.h>
#include <stdint.h>
#include <cstddef>
@@ -31,14 +29,12 @@ inline const __int128 MIN_INT128 = ((__int128)0x01 << 127);
class LargeIntValue {
public:
- static int64_t to_buffer(__int128 value, char* buffer) {
- return fmt::format_to(buffer, FMT_COMPILE("{}"), value) - buffer;
- }
-
- static std::string to_string(__int128 value) { return
fmt::format(FMT_COMPILE("{}"), value); }
- static std::string to_string(__uint128_t value) {
- return fmt::format(FMT_COMPILE("{}"), value);
- }
+ // Defined in large_int_value.cpp: instantiating the FMT_COMPILE int128
+ // formatter in every includer of this header is expensive.
+ static int64_t to_buffer(__int128 value, char* buffer);
+
+ static std::string to_string(__int128 value);
+ static std::string to_string(__uint128_t value);
};
std::ostream& operator<<(std::ostream& os, __int128 const& value);
diff --git a/be/src/core/wide_integer_from_double.cpp
b/be/src/core/wide_integer_from_double.cpp
new file mode 100644
index 00000000000..e0f8b480f39
--- /dev/null
+++ b/be/src/core/wide_integer_from_double.cpp
@@ -0,0 +1,50 @@
+// 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.
+
+// On platforms whose long double lacks an 80-bit mantissa (LDBL_MANT_DIG !=
64),
+// wide_integer_impl.h emulates the from-double intermediate type with
+// boost::multiprecision::cpp_bin_float_double_extended. Making that include
part
+// of every TU's closure costs ~4.3MB of preprocessed text; instead, ordinary
TUs
+// only see declarations of set_multiplier / wide_integer_from_builtin(double)
+// and this TU compiles the bodies once. The macro below switches
+// wide_integer_impl.h into "definitions inline here" mode.
+#define DORIS_WIDE_INTEGER_FROM_DOUBLE_IMPL_TU
+
+#include "core/wide_integer.h"
+
+// The `defined(...)` term below is what marks the macro above as used, and it
has to stay the
+// left operand: the preprocessor short-circuits `&&`, so a false left term
would skip it and the
+// definition would read as an unused macro (-Werror,-Wunused-macros). That
happens on exactly the
+// platforms this file exists for the inverse of -- where the 80-bit long
double exists,
+// wide_integer_impl.h takes its `#if (LDBL_MANT_DIG == 64)` branch and never
evaluates the
+// `#elif defined(...)` that would otherwise consume the macro. Reading it
here is also honest:
+// these instantiations exist precisely because this TU is the impl TU.
+#if defined(DORIS_WIDE_INTEGER_FROM_DOUBLE_IMPL_TU) && !(LDBL_MANT_DIG == 64)
+
+// Every wide::integer specialization used by Doris (see core/extended_types.h)
+// gets its from-double conversion emitted here. A new specialization that is
+// constructed from double elsewhere will fail at link time — add it below.
+template void wide::integer<128, signed>::_impl::wide_integer_from_builtin(
+ wide::integer<128, signed>&, double) noexcept;
+template void wide::integer<128, unsigned>::_impl::wide_integer_from_builtin(
+ wide::integer<128, unsigned>&, double) noexcept;
+template void wide::integer<256, signed>::_impl::wide_integer_from_builtin(
+ wide::integer<256, signed>&, double) noexcept;
+template void wide::integer<256, unsigned>::_impl::wide_integer_from_builtin(
+ wide::integer<256, unsigned>&, double) noexcept;
+
+#endif // !(LDBL_MANT_DIG == 64)
diff --git a/be/src/core/wide_integer_impl.h b/be/src/core/wide_integer_impl.h
index a5ad6ae041b..a6e9ecf8032 100644
--- a/be/src/core/wide_integer_impl.h
+++ b/be/src/core/wide_integer_impl.h
@@ -23,7 +23,6 @@
// and modified by Doris
#pragma once
-#include <boost/math/special_functions/fpclassify.hpp>
#include <cassert>
#include <cfloat>
#include <cmath>
@@ -37,13 +36,28 @@
/// Use same extended double for all platforms
#if (LDBL_MANT_DIG == 64)
+#include <boost/math/special_functions/fpclassify.hpp>
+
#define CONSTEXPR_FROM_DOUBLE constexpr
+#define DORIS_WIDE_FROM_DOUBLE_INLINE 1
using FromDoubleIntermediateType = long double;
-#else
+#elif defined(DORIS_WIDE_INTEGER_FROM_DOUBLE_IMPL_TU)
+#include <boost/math/special_functions/fpclassify.hpp>
#include <boost/multiprecision/cpp_bin_float.hpp>
/// `wide_integer_from_builtin` can't be constexpr with non-literal
`cpp_bin_float_double_extended`
#define CONSTEXPR_FROM_DOUBLE
+#define DORIS_WIDE_FROM_DOUBLE_INLINE 1
using FromDoubleIntermediateType =
boost::multiprecision::cpp_bin_float_double_extended;
+#else
+/// Platforms without an 80-bit long double emulate the intermediate type with
+/// boost::multiprecision, which drags ~4.3MB of preprocessed closure into
every
+/// TU. Ordinary TUs therefore see only declarations of the two from-double
+/// members below; the bodies are compiled once in
+/// core/wide_integer_from_double.cpp (which defines the *_IMPL_TU macro above
+/// and emits explicit instantiations). wide_integer_from_builtin(double) was
+/// never constexpr on these platforms, so no constant evaluation is lost.
+#define CONSTEXPR_FROM_DOUBLE
+#define DORIS_WIDE_FROM_DOUBLE_INLINE 0
#endif
namespace wide {
@@ -308,7 +322,11 @@ struct integer<Bits, Signed>::_impl {
* a_(n - 1) = a_n * max_int + b2, a_n <= max_int <- base case.
*/
template <class T>
- constexpr static void set_multiplier(integer<Bits, Signed>& self, T t)
noexcept {
+ CONSTEXPR_FROM_DOUBLE static void set_multiplier(integer<Bits, Signed>&
self, T t) noexcept
+#if !DORIS_WIDE_FROM_DOUBLE_INLINE
+ ;
+#else
+ {
constexpr uint64_t max_int = std::numeric_limits<uint64_t>::max();
static_assert(std::is_same_v<T, double> || std::is_same_v<T,
FromDoubleIntermediateType>);
/// Implementation specific behaviour on overflow (if we don't check
here, stack overflow will triggered in bigint_cast).
@@ -343,9 +361,14 @@ struct integer<Bits, Signed>::_impl {
self += static_cast<uint64_t>(t - floor(static_cast<double>(alpha)) *
static_cast<T>(max_int)); //
+= b_i
}
+#endif
CONSTEXPR_FROM_DOUBLE static void wide_integer_from_builtin(integer<Bits,
Signed>& self,
- double rhs)
noexcept {
+ double rhs)
noexcept
+#if !DORIS_WIDE_FROM_DOUBLE_INLINE
+ ;
+#else
+ {
constexpr int64_t max_int = std::numeric_limits<int64_t>::max();
constexpr int64_t min_int = std::numeric_limits<int64_t>::lowest();
@@ -374,6 +397,7 @@ struct integer<Bits, Signed>::_impl {
self = -self;
}
}
+#endif
template <size_t Bits2, typename Signed2>
constexpr static void wide_integer_from_wide_integer(
diff --git a/be/src/exec/common/join_op_utils.h
b/be/src/exec/common/join_op_utils.h
new file mode 100644
index 00000000000..7b6636f88dc
--- /dev/null
+++ b/be/src/exec/common/join_op_utils.h
@@ -0,0 +1,157 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#pragma once
+
+// Lightweight join-op level types split out of join_utils.h so that
+// exec/pipeline/dependency.h (which holds JoinOpVariants / AsofIndexVariant
+// members by value) does not have to see the hash-table machinery.
+// Everything here depends only on thrift enums and std containers.
+
+#include <gen_cpp/PlanNodes_types.h>
+#include <pdqsort.h>
+
+#include <cstdint>
+#include <variant>
+#include <vector>
+
+#include "common/compiler_util.h"
+
+namespace doris {
+
+using JoinOpVariants =
+ std::variant<std::integral_constant<TJoinOp::type,
TJoinOp::INNER_JOIN>,
+ std::integral_constant<TJoinOp::type,
TJoinOp::LEFT_SEMI_JOIN>,
+ std::integral_constant<TJoinOp::type,
TJoinOp::LEFT_ANTI_JOIN>,
+ std::integral_constant<TJoinOp::type,
TJoinOp::LEFT_OUTER_JOIN>,
+ std::integral_constant<TJoinOp::type,
TJoinOp::FULL_OUTER_JOIN>,
+ std::integral_constant<TJoinOp::type,
TJoinOp::RIGHT_OUTER_JOIN>,
+ std::integral_constant<TJoinOp::type,
TJoinOp::CROSS_JOIN>,
+ std::integral_constant<TJoinOp::type,
TJoinOp::RIGHT_SEMI_JOIN>,
+ std::integral_constant<TJoinOp::type,
TJoinOp::RIGHT_ANTI_JOIN>,
+ std::integral_constant<TJoinOp::type,
TJoinOp::NULL_AWARE_LEFT_ANTI_JOIN>,
+ std::integral_constant<TJoinOp::type,
TJoinOp::NULL_AWARE_LEFT_SEMI_JOIN>,
+ std::integral_constant<TJoinOp::type,
TJoinOp::ASOF_LEFT_INNER_JOIN>,
+ std::integral_constant<TJoinOp::type,
TJoinOp::ASOF_LEFT_OUTER_JOIN>>;
+
+inline bool is_asof_join(TJoinOp::type join_op) {
+ return join_op == TJoinOp::ASOF_LEFT_INNER_JOIN || join_op ==
TJoinOp::ASOF_LEFT_OUTER_JOIN;
+}
+
+template <int JoinOpType>
+inline constexpr bool is_asof_join_op_v =
+ JoinOpType == TJoinOp::ASOF_LEFT_INNER_JOIN || JoinOpType ==
TJoinOp::ASOF_LEFT_OUTER_JOIN;
+
+template <int JoinOpType>
+inline constexpr bool is_asof_outer_join_op_v = JoinOpType ==
TJoinOp::ASOF_LEFT_OUTER_JOIN;
+
+// ASOF JOIN index with inline values for cache-friendly branchless binary
search.
+// IntType is the integer representation of the ASOF column value:
+// uint32_t for DateV2, uint64_t for DateTimeV2 and TimestampTZ.
+// Rows are sorted by asof_value during build, then materialized into SoA
arrays
+// so probe-side binary search only touches the ASOF values hot path.
+template <typename IntType>
+struct AsofIndexGroup {
+ using int_type = IntType;
+
+ struct Entry {
+ IntType asof_value;
+ uint32_t row_index; // 1-based, 0 = invalid/padding
+ };
+
+ std::vector<Entry> entries;
+ std::vector<IntType> asof_values;
+ std::vector<uint32_t> row_indexes;
+
+ void add_row(IntType value, uint32_t row_idx) { entries.push_back({value,
row_idx}); }
+
+ void sort_and_finalize() {
+ if (entries.empty()) {
+ return;
+ }
+ if (entries.size() > 1) {
+ pdqsort(entries.begin(), entries.end(),
+ [](const Entry& a, const Entry& b) { return a.asof_value <
b.asof_value; });
+ }
+
+ asof_values.resize(entries.size());
+ row_indexes.resize(entries.size());
+ for (size_t i = 0; i < entries.size(); ++i) {
+ asof_values[i] = entries[i].asof_value;
+ row_indexes[i] = entries[i].row_index;
+ }
+
+ std::vector<Entry>().swap(entries);
+ }
+
+ const IntType* values_data() const { return asof_values.data(); }
+
+ // Branchless lower_bound: first i where asof_values[i] >= target
+ ALWAYS_INLINE size_t lower_bound(IntType target) const {
+ size_t lo = 0, n = asof_values.size();
+ while (n > 1) {
+ size_t half = n / 2;
+ lo += half * (asof_values[lo + half] < target);
+ n -= half;
+ }
+ if (lo < asof_values.size()) {
+ lo += (asof_values[lo] < target);
+ }
+ return lo;
+ }
+
+ // Branchless upper_bound: first i where asof_values[i] > target
+ ALWAYS_INLINE size_t upper_bound(IntType target) const {
+ size_t lo = 0, n = asof_values.size();
+ while (n > 1) {
+ size_t half = n / 2;
+ lo += half * (asof_values[lo + half] <= target);
+ n -= half;
+ }
+ if (lo < asof_values.size()) {
+ lo += (asof_values[lo] <= target);
+ }
+ return lo;
+ }
+
+ // Semantics by (is_greater, is_strict):
+ // (true, false): probe >= build -> find largest build value <= probe
+ // (true, true): probe > build -> find largest build value < probe
+ // (false, false): probe <= build -> find smallest build value >= probe
+ // (false, true): probe < build -> find smallest build value > probe
+ // Returns the build row index of the best match, or 0 if no match.
+ template <bool IsGreater, bool IsStrict>
+ ALWAYS_INLINE uint32_t find_best_match(IntType probe_value) const {
+ if (asof_values.empty()) {
+ return 0;
+ }
+ if constexpr (IsGreater) {
+ size_t pos = IsStrict ? lower_bound(probe_value) :
upper_bound(probe_value);
+ return pos > 0 ? row_indexes[pos - 1] : 0;
+ } else {
+ size_t pos = IsStrict ? upper_bound(probe_value) :
lower_bound(probe_value);
+ return pos < asof_values.size() ? row_indexes[pos] : 0;
+ }
+ }
+};
+
+// Type-erased container for all ASOF index groups.
+// DateV2 -> uint32_t, DateTimeV2/TimestampTZ -> uint64_t.
+using AsofIndexVariant = std::variant<std::monostate,
std::vector<AsofIndexGroup<uint32_t>>,
+ std::vector<AsofIndexGroup<uint64_t>>>;
+
+} // namespace doris
diff --git a/be/src/exec/common/join_utils.h b/be/src/exec/common/join_utils.h
index 7482c6fbdae..fc0ae935adc 100644
--- a/be/src/exec/common/join_utils.h
+++ b/be/src/exec/common/join_utils.h
@@ -24,6 +24,7 @@
#include "exec/common/hash_table/hash_key_type.h"
#include "exec/common/hash_table/hash_map_context.h"
#include "exec/common/hash_table/join_hash_table.h"
+#include "exec/common/join_op_utils.h" // IWYU pragma: export
(JoinOpVariants/AsofIndexGroup moved there)
namespace doris {
@@ -45,32 +46,6 @@ decltype(auto) asof_column_dispatch(const IColumn* col,
Func&& func) {
return std::forward<Func>(func)(col);
}
}
-using JoinOpVariants =
- std::variant<std::integral_constant<TJoinOp::type,
TJoinOp::INNER_JOIN>,
- std::integral_constant<TJoinOp::type,
TJoinOp::LEFT_SEMI_JOIN>,
- std::integral_constant<TJoinOp::type,
TJoinOp::LEFT_ANTI_JOIN>,
- std::integral_constant<TJoinOp::type,
TJoinOp::LEFT_OUTER_JOIN>,
- std::integral_constant<TJoinOp::type,
TJoinOp::FULL_OUTER_JOIN>,
- std::integral_constant<TJoinOp::type,
TJoinOp::RIGHT_OUTER_JOIN>,
- std::integral_constant<TJoinOp::type,
TJoinOp::CROSS_JOIN>,
- std::integral_constant<TJoinOp::type,
TJoinOp::RIGHT_SEMI_JOIN>,
- std::integral_constant<TJoinOp::type,
TJoinOp::RIGHT_ANTI_JOIN>,
- std::integral_constant<TJoinOp::type,
TJoinOp::NULL_AWARE_LEFT_ANTI_JOIN>,
- std::integral_constant<TJoinOp::type,
TJoinOp::NULL_AWARE_LEFT_SEMI_JOIN>,
- std::integral_constant<TJoinOp::type,
TJoinOp::ASOF_LEFT_INNER_JOIN>,
- std::integral_constant<TJoinOp::type,
TJoinOp::ASOF_LEFT_OUTER_JOIN>>;
-
-inline bool is_asof_join(TJoinOp::type join_op) {
- return join_op == TJoinOp::ASOF_LEFT_INNER_JOIN || join_op ==
TJoinOp::ASOF_LEFT_OUTER_JOIN;
-}
-
-template <int JoinOpType>
-inline constexpr bool is_asof_join_op_v =
- JoinOpType == TJoinOp::ASOF_LEFT_INNER_JOIN || JoinOpType ==
TJoinOp::ASOF_LEFT_OUTER_JOIN;
-
-template <int JoinOpType>
-inline constexpr bool is_asof_outer_join_op_v = JoinOpType ==
TJoinOp::ASOF_LEFT_OUTER_JOIN;
-
template <class T>
using PrimaryTypeHashTableContext = MethodOneNumber<T, JoinHashMap<T,
HashCRC32Return32<T>, false>>;
@@ -232,99 +207,4 @@ inline void try_convert_to_direct_mapping(
primary_to_direct_mapping(context, key_columns, variant_ptrs);
}
-// ASOF JOIN index with inline values for cache-friendly branchless binary
search.
-// IntType is the integer representation of the ASOF column value:
-// uint32_t for DateV2, uint64_t for DateTimeV2 and TimestampTZ.
-// Rows are sorted by asof_value during build, then materialized into SoA
arrays
-// so probe-side binary search only touches the ASOF values hot path.
-template <typename IntType>
-struct AsofIndexGroup {
- using int_type = IntType;
-
- struct Entry {
- IntType asof_value;
- uint32_t row_index; // 1-based, 0 = invalid/padding
- };
-
- std::vector<Entry> entries;
- std::vector<IntType> asof_values;
- std::vector<uint32_t> row_indexes;
-
- void add_row(IntType value, uint32_t row_idx) { entries.push_back({value,
row_idx}); }
-
- void sort_and_finalize() {
- if (entries.empty()) {
- return;
- }
- if (entries.size() > 1) {
- pdqsort(entries.begin(), entries.end(),
- [](const Entry& a, const Entry& b) { return a.asof_value <
b.asof_value; });
- }
-
- asof_values.resize(entries.size());
- row_indexes.resize(entries.size());
- for (size_t i = 0; i < entries.size(); ++i) {
- asof_values[i] = entries[i].asof_value;
- row_indexes[i] = entries[i].row_index;
- }
-
- std::vector<Entry>().swap(entries);
- }
-
- const IntType* values_data() const { return asof_values.data(); }
-
- // Branchless lower_bound: first i where asof_values[i] >= target
- ALWAYS_INLINE size_t lower_bound(IntType target) const {
- size_t lo = 0, n = asof_values.size();
- while (n > 1) {
- size_t half = n / 2;
- lo += half * (asof_values[lo + half] < target);
- n -= half;
- }
- if (lo < asof_values.size()) {
- lo += (asof_values[lo] < target);
- }
- return lo;
- }
-
- // Branchless upper_bound: first i where asof_values[i] > target
- ALWAYS_INLINE size_t upper_bound(IntType target) const {
- size_t lo = 0, n = asof_values.size();
- while (n > 1) {
- size_t half = n / 2;
- lo += half * (asof_values[lo + half] <= target);
- n -= half;
- }
- if (lo < asof_values.size()) {
- lo += (asof_values[lo] <= target);
- }
- return lo;
- }
-
- // Semantics by (is_greater, is_strict):
- // (true, false): probe >= build -> find largest build value <= probe
- // (true, true): probe > build -> find largest build value < probe
- // (false, false): probe <= build -> find smallest build value >= probe
- // (false, true): probe < build -> find smallest build value > probe
- // Returns the build row index of the best match, or 0 if no match.
- template <bool IsGreater, bool IsStrict>
- ALWAYS_INLINE uint32_t find_best_match(IntType probe_value) const {
- if (asof_values.empty()) {
- return 0;
- }
- if constexpr (IsGreater) {
- size_t pos = IsStrict ? lower_bound(probe_value) :
upper_bound(probe_value);
- return pos > 0 ? row_indexes[pos - 1] : 0;
- } else {
- size_t pos = IsStrict ? upper_bound(probe_value) :
lower_bound(probe_value);
- return pos < asof_values.size() ? row_indexes[pos] : 0;
- }
- }
-};
-
-// Type-erased container for all ASOF index groups.
-// DateV2 -> uint32_t, DateTimeV2/TimestampTZ -> uint64_t.
-using AsofIndexVariant = std::variant<std::monostate,
std::vector<AsofIndexGroup<uint32_t>>,
- std::vector<AsofIndexGroup<uint64_t>>>;
-
} // namespace doris
diff --git a/be/src/exec/exchange/local_exchanger.h
b/be/src/exec/exchange/local_exchanger.h
index e0a821a3643..f1bb9f11052 100644
--- a/be/src/exec/exchange/local_exchanger.h
+++ b/be/src/exec/exchange/local_exchanger.h
@@ -17,6 +17,8 @@
#pragma once
+#include <concurrentqueue.h>
+
#include "exec/operator/operator.h"
#include "exec/pipeline/dependency.h"
diff --git a/be/src/exec/operator/aggregation_sink_operator.h
b/be/src/exec/operator/aggregation_sink_operator.h
index 50b32ecb364..28f1845e3c9 100644
--- a/be/src/exec/operator/aggregation_sink_operator.h
+++ b/be/src/exec/operator/aggregation_sink_operator.h
@@ -19,6 +19,7 @@
#include <stdint.h>
+#include "exec/common/agg_utils.h"
#include "exec/operator/operator.h"
#include "runtime/exec_env.h"
#include "runtime/runtime_profile.h"
diff --git a/be/src/exec/operator/aggregation_source_operator.cpp
b/be/src/exec/operator/aggregation_source_operator.cpp
index f90ebd9bb47..f474c09b6b1 100644
--- a/be/src/exec/operator/aggregation_source_operator.cpp
+++ b/be/src/exec/operator/aggregation_source_operator.cpp
@@ -22,6 +22,7 @@
#include "common/exception.h"
#include "core/column/column_fixed_length_object.h"
+#include "exec/common/agg_utils.h"
#include "exec/operator/operator.h"
#include "exprs/vectorized_agg_fn.h"
#include "exprs/vexpr_fwd.h"
diff --git a/be/src/exec/operator/bucketed_aggregation_sink_operator.h
b/be/src/exec/operator/bucketed_aggregation_sink_operator.h
index 891977b226c..5c56b65efd1 100644
--- a/be/src/exec/operator/bucketed_aggregation_sink_operator.h
+++ b/be/src/exec/operator/bucketed_aggregation_sink_operator.h
@@ -19,6 +19,7 @@
#include <stdint.h>
+#include "exec/common/agg_utils.h"
#include "exec/operator/operator.h"
#include "runtime/exec_env.h"
#include "runtime/runtime_profile.h"
diff --git a/be/src/exec/operator/bucketed_aggregation_source_operator.cpp
b/be/src/exec/operator/bucketed_aggregation_source_operator.cpp
index ba0a1b650bc..a9ff2db49c4 100644
--- a/be/src/exec/operator/bucketed_aggregation_source_operator.cpp
+++ b/be/src/exec/operator/bucketed_aggregation_source_operator.cpp
@@ -22,6 +22,7 @@
#include "common/exception.h"
#include "core/column/column_vector.h"
+#include "exec/common/agg_utils.h"
#include "exec/common/hash_table/hash.h"
#include "exec/common/util.hpp"
#include "exec/operator/operator.h"
diff --git a/be/src/exec/operator/hashjoin_build_sink.cpp
b/be/src/exec/operator/hashjoin_build_sink.cpp
index 55716412701..2d0aea94ffc 100644
--- a/be/src/exec/operator/hashjoin_build_sink.cpp
+++ b/be/src/exec/operator/hashjoin_build_sink.cpp
@@ -25,6 +25,7 @@
#include "core/column/column_const.h"
#include "core/column/column_nullable.h"
#include "core/data_type/data_type_nullable.h"
+#include "exec/common/hash_table/hash_map_util.h"
#include "exec/common/template_helpers.hpp"
#include "exec/operator/hashjoin_probe_operator.h"
#include "exec/operator/operator.h"
diff --git a/be/src/exec/operator/hashjoin_probe_operator.cpp
b/be/src/exec/operator/hashjoin_probe_operator.cpp
index 44d9e806672..4214928af4b 100644
--- a/be/src/exec/operator/hashjoin_probe_operator.cpp
+++ b/be/src/exec/operator/hashjoin_probe_operator.cpp
@@ -28,6 +28,7 @@
#include "core/column/column_const.h"
#include "core/column/column_nullable.h"
#include "core/data_type/data_type_nullable.h"
+#include "exec/common/join_utils.h"
#include "exec/operator/operator.h"
#include "runtime/descriptors.h"
#include "util/uid_util.h"
diff --git a/be/src/exec/operator/hashjoin_probe_operator.h
b/be/src/exec/operator/hashjoin_probe_operator.h
index 371a43fe1eb..2542efe9c88 100644
--- a/be/src/exec/operator/hashjoin_probe_operator.h
+++ b/be/src/exec/operator/hashjoin_probe_operator.h
@@ -20,6 +20,7 @@
#include "common/be_mock_util.h"
#include "common/status.h"
+#include "exec/operator/join/process_hash_table_probe.h"
#include "exec/operator/join_probe_operator.h"
#include "exec/operator/operator.h"
diff --git a/be/src/exec/operator/join/process_hash_table_probe.h
b/be/src/exec/operator/join/process_hash_table_probe.h
index f17e49d448f..f39e865f8cd 100644
--- a/be/src/exec/operator/join/process_hash_table_probe.h
+++ b/be/src/exec/operator/join/process_hash_table_probe.h
@@ -23,6 +23,7 @@
#include "core/column/column.h"
#include "core/column/column_vector.h"
#include "core/custom_allocator.h"
+#include "runtime/runtime_profile.h"
namespace doris {
class Block;
diff --git a/be/src/exec/operator/join/process_hash_table_probe_impl.h
b/be/src/exec/operator/join/process_hash_table_probe_impl.h
index 03401a089f3..e3713f8a00f 100644
--- a/be/src/exec/operator/join/process_hash_table_probe_impl.h
+++ b/be/src/exec/operator/join/process_hash_table_probe_impl.h
@@ -25,6 +25,7 @@
#include "core/column/column_filter_helper.h"
#include "core/column/column_nullable.h"
#include "core/data_type/data_type_number.h" // IWYU pragma: keep
+#include "exec/common/join_utils.h"
#include "exec/operator/hashjoin_probe_operator.h"
#include "exec/operator/join/process_hash_table_probe.h"
#include "exprs/vexpr_context.h"
diff --git a/be/src/exec/operator/join_build_sink_operator.h
b/be/src/exec/operator/join_build_sink_operator.h
index 006d6c1f551..9fc27e471bc 100644
--- a/be/src/exec/operator/join_build_sink_operator.h
+++ b/be/src/exec/operator/join_build_sink_operator.h
@@ -17,6 +17,7 @@
#pragma once
+#include "exec/common/join_utils.h"
#include "exec/operator/operator.h"
namespace doris {
diff --git a/be/src/exec/operator/materialization_opertor.h
b/be/src/exec/operator/materialization_opertor.h
index bc6618b1bb0..7dfdf69f823 100644
--- a/be/src/exec/operator/materialization_opertor.h
+++ b/be/src/exec/operator/materialization_opertor.h
@@ -25,6 +25,7 @@
#include "common/status.h"
#include "exec/operator/operator.h"
+#include "service/brpc.h" // IWYU pragma: keep (brpc::Controller member below)
namespace doris {
class RuntimeState;
diff --git a/be/src/exec/operator/partitioned_aggregation_source_operator.cpp
b/be/src/exec/operator/partitioned_aggregation_source_operator.cpp
index 14acb727052..58018b6a529 100644
--- a/be/src/exec/operator/partitioned_aggregation_source_operator.cpp
+++ b/be/src/exec/operator/partitioned_aggregation_source_operator.cpp
@@ -25,6 +25,7 @@
#include "common/exception.h"
#include "common/logging.h"
#include "common/status.h"
+#include "exec/common/agg_utils.h"
#include "exec/operator/aggregation_source_operator.h"
#include "exec/operator/operator.h"
#include "exec/operator/spill_utils.h"
diff --git a/be/src/exec/operator/rec_cte_source_operator.cpp
b/be/src/exec/operator/rec_cte_source_operator.cpp
index 1c32983b22a..63331bbdfbc 100644
--- a/be/src/exec/operator/rec_cte_source_operator.cpp
+++ b/be/src/exec/operator/rec_cte_source_operator.cpp
@@ -17,6 +17,8 @@
#include "exec/operator/rec_cte_source_operator.h"
+#include "exec/common/distinct_agg_utils.h"
+
namespace doris {
RecCTESourceLocalState::RecCTESourceLocalState(RuntimeState* state,
OperatorXBase* parent)
diff --git a/be/src/exec/operator/set_probe_sink_operator.cpp
b/be/src/exec/operator/set_probe_sink_operator.cpp
index c3fb08a3725..4d77bd22bc8 100644
--- a/be/src/exec/operator/set_probe_sink_operator.cpp
+++ b/be/src/exec/operator/set_probe_sink_operator.cpp
@@ -22,6 +22,7 @@
#include <memory>
#include "exec/common/hash_table/hash_table_set_probe.h"
+#include "exec/common/set_utils.h"
#include "exec/operator/operator.h"
#include "exec/pipeline/pipeline_task.h"
diff --git a/be/src/exec/operator/set_sink_operator.cpp
b/be/src/exec/operator/set_sink_operator.cpp
index 614a45afbc8..b90c12bb994 100644
--- a/be/src/exec/operator/set_sink_operator.cpp
+++ b/be/src/exec/operator/set_sink_operator.cpp
@@ -21,6 +21,7 @@
#include "core/block/materialize_block.h"
#include "exec/common/hash_table/hash_table_set_build.h"
+#include "exec/common/set_utils.h"
#include "exec/operator/operator.h"
namespace doris {
diff --git a/be/src/exec/operator/set_source_operator.cpp
b/be/src/exec/operator/set_source_operator.cpp
index 63afdae0814..7f6394e20e6 100644
--- a/be/src/exec/operator/set_source_operator.cpp
+++ b/be/src/exec/operator/set_source_operator.cpp
@@ -21,6 +21,7 @@
#include <type_traits>
#include "common/status.h"
+#include "exec/common/set_utils.h"
#include "exec/operator/operator.h"
#include "runtime/runtime_profile.h"
diff --git a/be/src/exec/operator/streaming_aggregation_operator.h
b/be/src/exec/operator/streaming_aggregation_operator.h
index b23c79477a1..1c57d0ca33f 100644
--- a/be/src/exec/operator/streaming_aggregation_operator.h
+++ b/be/src/exec/operator/streaming_aggregation_operator.h
@@ -23,6 +23,7 @@
#include "common/status.h"
#include "core/block/block.h"
+#include "exec/common/agg_utils.h"
#include "exec/operator/operator.h"
#include "runtime/runtime_profile.h"
diff --git a/be/src/exec/pipeline/dependency.cpp
b/be/src/exec/pipeline/dependency.cpp
index 3f50bb6e977..f808e6b10e8 100644
--- a/be/src/exec/pipeline/dependency.cpp
+++ b/be/src/exec/pipeline/dependency.cpp
@@ -21,6 +21,11 @@
#include <mutex>
#include "common/logging.h"
+#include "exec/common/agg_utils.h"
+#include "exec/common/hash_table/hash_map_util.h"
+#include "exec/common/join_utils.h"
+#include "exec/common/set_utils.h"
+#include "exec/common/template_helpers.hpp"
#include "exec/operator/multi_cast_data_streamer.h"
#include "exec/pipeline/pipeline_fragment_context.h"
#include "exec/pipeline/pipeline_task.h"
@@ -197,6 +202,133 @@ LocalExchangeSharedState::LocalExchangeSharedState(int
num_instances) {
mem_counters.resize(num_instances, nullptr);
}
+AggSharedState::AggSharedState() {
+ agg_data = std::make_unique<AggregatedDataVariants>();
+}
+
+AggSharedState::~AggSharedState() {
+ if (!probe_expr_ctxs.empty()) {
+ _close_with_serialized_key();
+ } else {
+ _close_without_key();
+ }
+}
+
+void AggSharedState::_close_with_serialized_key() {
+ std::visit(Overload {[&](std::monostate& arg) -> void {
+ // Do nothing
+ },
+ [&](auto& agg_method) -> void {
+ if (use_simple_count) {
+ // Inline count: mapped slots hold UInt64,
+ // not real agg state pointers. Skip destroy.
+ return;
+ }
+ auto& data = *agg_method.hash_table;
+ data.for_each_mapped([&](auto& mapped) {
+ if (mapped) {
+ _destroy_agg_status(mapped);
+ mapped = nullptr;
+ }
+ });
+ if (data.has_null_key_data()) {
+ _destroy_agg_status(
+ data.template
get_null_key_data<AggregateDataPtr>());
+ }
+ }},
+ agg_data->method_variant);
+}
+
+void AggSharedState::_close_without_key() {
+ //because prepare maybe failed, and couldn't create agg data.
+ //but finally call close to destory agg data, if agg data has bitmapValue
+ //will be core dump, it's not initialized
+ if (agg_data_created_without_key) {
+ _destroy_agg_status(agg_data->without_key);
+ agg_data_created_without_key = false;
+ }
+}
+
+BucketedAggSharedState::PerInstanceData::PerInstanceData() :
arena(std::make_unique<Arena>()) {
+ bucket_agg_data.resize(BUCKETED_AGG_NUM_BUCKETS);
+ for (auto& p : bucket_agg_data) {
+ p = std::make_unique<BucketedAggDataVariants>();
+ }
+}
+
+BucketedAggSharedState::~BucketedAggSharedState() {
+ _close();
+}
+
+Status BucketedAggSharedState::init_instances(int num_instances,
+ const std::function<Status()>&
metadata_init) {
+ std::call_once(_init_once, [&]() {
+ num_sink_instances = num_instances;
+ per_instance_data.resize(num_instances);
+ sink_finished = std::make_unique<std::atomic<bool>[]>(num_instances);
+ for (int i = 0; i < num_instances; ++i) {
+ sink_finished[i].store(false, std::memory_order_relaxed);
+ }
+ for (auto& bs : bucket_states) {
+ bs.merged_instances.resize(num_instances, false);
+ }
+ _init_status = metadata_init();
+ });
+ return _init_status;
+}
+
+void BucketedAggSharedState::_close() {
+ for (auto& inst : per_instance_data) {
+ for (auto& bucket_data : inst.bucket_agg_data) {
+ _close_one_agg_data(*bucket_data);
+ }
+ }
+}
+
+void BucketedAggSharedState::_close_one_agg_data(BucketedAggDataVariants&
agg_data) {
+ std::visit(Overload {[&](std::monostate& arg) -> void {
+ // Do nothing
+ },
+ [&](auto& agg_method) -> void {
+ if (use_simple_count) {
+ // simple_count: mapped slots hold UInt64
counters,
+ // not real agg state pointers. Skip destroy.
+ return;
+ }
+ auto& data = *agg_method.hash_table;
+ data.for_each_mapped([&](auto& mapped) {
+ if (mapped) {
+ _destroy_agg_status(mapped);
+ mapped = nullptr;
+ }
+ });
+ if constexpr
(std::is_assignable_v<decltype(data.has_null_key_data()),
+ bool>) {
+ if (data.has_null_key_data()) {
+ _destroy_agg_status(
+ data.template
get_null_key_data<AggregateDataPtr>());
+ }
+ }
+ }},
+ agg_data.method_variant);
+}
+
+HashJoinSharedState::HashJoinSharedState() {
+ hash_table_variant_vector.push_back(std::make_shared<JoinDataVariants>());
+}
+
+HashJoinSharedState::HashJoinSharedState(int num_instances) {
+ source_deps.resize(num_instances, nullptr);
+ hash_table_variant_vector.resize(num_instances, nullptr);
+ for (int i = 0; i < num_instances; i++) {
+ hash_table_variant_vector[i] = std::make_shared<JoinDataVariants>();
+ }
+}
+
+SetSharedState::SetSharedState() :
hash_table_variants(std::make_unique<SetDataVariants>()) {}
+
+SetSharedState::~SetSharedState() = default;
+
MutableColumns AggSharedState::_get_keys_hash_table() {
return std::visit(
Overload {[&](std::monostate& arg) {
diff --git a/be/src/exec/pipeline/dependency.h
b/be/src/exec/pipeline/dependency.h
index a8b7ad4402b..53f9ed9281b 100644
--- a/be/src/exec/pipeline/dependency.h
+++ b/be/src/exec/pipeline/dependency.h
@@ -22,7 +22,6 @@
#include <sys/_types/_u_int.h>
#endif
-#include <concurrentqueue.h>
#include <gen_cpp/Partitions_types.h>
#include <gen_cpp/internal_service.pb.h>
#include <sqltypes.h>
@@ -30,31 +29,44 @@
#include <atomic>
#include <condition_variable>
#include <functional>
+#include <list>
#include <memory>
#include <mutex>
+#include <queue>
+#include <set>
#include <thread>
#include <utility>
+#include "common/cast_set.h"
#include "common/config.h"
+#include "common/exception.h"
+#include "common/factory_creator.h"
#include "common/logging.h"
#include "common/thread_safety_annotations.h"
+#include "core/arena.h"
#include "core/block/block.h"
#include "core/types.h"
-#include "exec/common/agg_utils.h"
-#include "exec/common/join_utils.h"
-#include "exec/common/set_utils.h"
+#include "exec/common/join_op_utils.h"
#include "exec/operator/data_queue.h"
-#include "exec/operator/join/process_hash_table_probe.h"
#include "exec/sort/partition_sorter.h"
#include "exec/sort/sorter.h"
#include "exec/spill/spill_file.h"
+#include "exprs/vexpr_fwd.h"
#include "runtime/runtime_profile_counter_names.h"
-#include "util/brpc_closure.h"
#include "util/stack_util.h"
+#include "util/stopwatch.hpp"
namespace doris {
class AggFnEvaluator;
class VSlotRef;
+// Heavy hash-table variant machinery (exec/common/*_utils.h) is referenced
+// through pointers only; keep it out of this header's parse cost.
+struct AggregatedDataVariants;
+struct AggregateDataContainer;
+struct BucketedAggDataVariants;
+struct JoinDataVariants;
+struct SetDataVariants;
+using AggregateDataPtr = char*;
} // namespace doris
namespace doris {
@@ -293,14 +305,11 @@ struct RuntimeFilterTimerQueue {
struct AggSharedState : public BasicSharedState {
ENABLE_FACTORY_CREATOR(AggSharedState)
public:
- AggSharedState() { agg_data = std::make_unique<AggregatedDataVariants>(); }
- ~AggSharedState() override {
- if (!probe_expr_ctxs.empty()) {
- _close_with_serialized_key();
- } else {
- _close_without_key();
- }
- }
+ // Defined in dependency.cpp: the bodies touch AggregatedDataVariants'
+ // full variant machinery, which every includer would otherwise
+ // instantiate at parse time.
+ AggSharedState();
+ ~AggSharedState() override;
Status reset_hash_table();
@@ -312,7 +321,7 @@ public:
// 2nd phase: is_merge=false, maybe have multiple exprs.
static int get_slot_column_id(const AggFnEvaluator* evaluator);
- AggregatedDataVariantsUPtr agg_data = nullptr;
+ std::unique_ptr<AggregatedDataVariants> agg_data;
std::unique_ptr<AggregateDataContainer> aggregate_data_container;
std::vector<AggFnEvaluator*> aggregate_evaluators;
// group by k1,k2
@@ -323,7 +332,7 @@ public:
size_t total_size_of_aggregate_states = 0;
size_t align_aggregate_states = 1;
/// The offset to the n-th aggregate function in a row of aggregate
functions.
- Sizes offsets_of_aggregate_states;
+ std::vector<size_t> offsets_of_aggregate_states;
std::vector<size_t> make_nullable_keys;
bool agg_data_created_without_key = false;
@@ -398,40 +407,8 @@ public:
private:
MutableColumns _get_keys_hash_table();
- void _close_with_serialized_key() {
- std::visit(Overload {[&](std::monostate& arg) -> void {
- // Do nothing
- },
- [&](auto& agg_method) -> void {
- if (use_simple_count) {
- // Inline count: mapped slots hold UInt64,
- // not real agg state pointers. Skip
destroy.
- return;
- }
- auto& data = *agg_method.hash_table;
- data.for_each_mapped([&](auto& mapped) {
- if (mapped) {
- _destroy_agg_status(mapped);
- mapped = nullptr;
- }
- });
- if (data.has_null_key_data()) {
- _destroy_agg_status(
- data.template
get_null_key_data<AggregateDataPtr>());
- }
- }},
- agg_data->method_variant);
- }
-
- void _close_without_key() {
- //because prepare maybe failed, and couldn't create agg data.
- //but finally call close to destory agg data, if agg data has
bitmapValue
- //will be core dump, it's not initialized
- if (agg_data_created_without_key) {
- _destroy_agg_status(agg_data->without_key);
- agg_data_created_without_key = false;
- }
- }
+ void _close_with_serialized_key();
+ void _close_without_key();
void _destroy_agg_status(AggregateDataPtr data);
};
@@ -460,22 +437,17 @@ struct BucketedAggSharedState : public BasicSharedState {
ENABLE_FACTORY_CREATOR(BucketedAggSharedState)
public:
BucketedAggSharedState() = default;
- ~BucketedAggSharedState() override { _close(); }
+ ~BucketedAggSharedState() override; // defined in dependency.cpp (variant
machinery)
/// Per-instance data. One per sink pipeline instance.
/// Each instance has 256 bucket hash tables + 1 shared arena.
struct PerInstanceData {
/// 256 per-bucket hash tables. Each bucket has its own
BucketedAggDataVariants.
/// Uses PHHashMap<StringRef> for string keys instead of StringHashMap.
- std::vector<BucketedAggDataVariantsUPtr> bucket_agg_data;
- ArenaUPtr arena;
+ std::vector<std::unique_ptr<BucketedAggDataVariants>> bucket_agg_data;
+ std::unique_ptr<Arena> arena;
- PerInstanceData() : arena(std::make_unique<Arena>()) {
- bucket_agg_data.resize(BUCKETED_AGG_NUM_BUCKETS);
- for (auto& p : bucket_agg_data) {
- p = std::make_unique<BucketedAggDataVariants>();
- }
- }
+ PerInstanceData(); // defined in dependency.cpp (variant machinery)
};
/// Per-bucket merge state for pipelined source-side processing.
@@ -514,7 +486,7 @@ public:
VExprContextSPtrs probe_expr_ctxs;
size_t total_size_of_aggregate_states = 0;
size_t align_aggregate_states = 1;
- Sizes offsets_of_aggregate_states;
+ std::vector<size_t> offsets_of_aggregate_states;
std::vector<size_t> make_nullable_keys;
std::atomic<size_t> input_num_rows {0};
@@ -542,64 +514,17 @@ public:
/// The callback runs exactly once (under std::call_once), must return
Status,
/// and should populate shared metadata like probe_expr_ctxs,
aggregate_evaluators, etc.
/// All threads observe the same init status via _init_status.
- template <typename Func>
- Status init_instances(int num_instances, Func&& metadata_init) {
- std::call_once(_init_once, [&]() {
- num_sink_instances = num_instances;
- per_instance_data.resize(num_instances);
- sink_finished =
std::make_unique<std::atomic<bool>[]>(num_instances);
- for (int i = 0; i < num_instances; ++i) {
- sink_finished[i].store(false, std::memory_order_relaxed);
- }
- for (auto& bs : bucket_states) {
- bs.merged_instances.resize(num_instances, false);
- }
- _init_status = std::forward<Func>(metadata_init)();
- });
- return _init_status;
- }
+ /// Once-per-query cold path; defined in dependency.cpp so the body (which
+ /// materializes the per-bucket BucketedAggDataVariants storage) stays out
+ /// of every includer's parse.
+ Status init_instances(int num_instances, const std::function<Status()>&
metadata_init);
private:
std::once_flag _init_once;
Status _init_status;
- void _close() {
- for (auto& inst : per_instance_data) {
- for (auto& bucket_data : inst.bucket_agg_data) {
- _close_one_agg_data(*bucket_data);
- }
- }
- }
-
- void _close_one_agg_data(BucketedAggDataVariants& agg_data) {
- std::visit(
- Overload {[&](std::monostate& arg) -> void {
- // Do nothing
- },
- [&](auto& agg_method) -> void {
- if (use_simple_count) {
- // simple_count: mapped slots hold UInt64
counters,
- // not real agg state pointers. Skip destroy.
- return;
- }
- auto& data = *agg_method.hash_table;
- data.for_each_mapped([&](auto& mapped) {
- if (mapped) {
- _destroy_agg_status(mapped);
- mapped = nullptr;
- }
- });
- if constexpr
(std::is_assignable_v<decltype(data.has_null_key_data()),
- bool>) {
- if (data.has_null_key_data()) {
- _destroy_agg_status(
- data.template
get_null_key_data<AggregateDataPtr>());
- }
- }
- }},
- agg_data.method_variant);
- }
-
+ void _close();
+ void _close_one_agg_data(BucketedAggDataVariants& agg_data);
void _destroy_agg_status(AggregateDataPtr data);
};
@@ -712,16 +637,9 @@ struct JoinSharedState : public BasicSharedState {
struct HashJoinSharedState : public JoinSharedState {
ENABLE_FACTORY_CREATOR(HashJoinSharedState)
- HashJoinSharedState() {
-
hash_table_variant_vector.push_back(std::make_shared<JoinDataVariants>());
- }
- HashJoinSharedState(int num_instances) {
- source_deps.resize(num_instances, nullptr);
- hash_table_variant_vector.resize(num_instances, nullptr);
- for (int i = 0; i < num_instances; i++) {
- hash_table_variant_vector[i] =
std::make_shared<JoinDataVariants>();
- }
- }
+ // Defined in dependency.cpp (they materialize JoinDataVariants).
+ HashJoinSharedState();
+ HashJoinSharedState(int num_instances);
std::shared_ptr<Arena> arena = std::make_shared<Arena>();
const std::vector<TupleDescriptor*> build_side_child_desc;
@@ -794,6 +712,11 @@ public:
struct SetSharedState : public BasicSharedState {
ENABLE_FACTORY_CREATOR(SetSharedState)
public:
+ // Defined in dependency.cpp: constructing/destroying SetDataVariants
+ // needs the full variant machinery.
+ SetSharedState();
+ ~SetSharedState() override;
+
/// default init
Block build_block; // build to source
//record element size in hashtable
@@ -804,9 +727,8 @@ public:
//// shared static states (shared, decided in prepare/open...)
- /// init in setup_local_state
- std::unique_ptr<SetDataVariants> hash_table_variants =
- std::make_unique<SetDataVariants>(); // the real data HERE.
+ /// init in setup_local_state (allocated in the constructor)
+ std::unique_ptr<SetDataVariants> hash_table_variants; // the real data
HERE.
std::vector<bool> build_not_ignore_null;
// The SET operator's child might have different nullable attributes.
diff --git a/be/src/exec/pipeline/rec_cte_shared_state.cpp
b/be/src/exec/pipeline/rec_cte_shared_state.cpp
new file mode 100644
index 00000000000..b0c41fd86ae
--- /dev/null
+++ b/be/src/exec/pipeline/rec_cte_shared_state.cpp
@@ -0,0 +1,157 @@
+// 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 "exec/pipeline/rec_cte_shared_state.h"
+
+#include <gen_cpp/internal_service.pb.h>
+
+#include "exec/common/distinct_agg_utils.h"
+#include "exec/common/template_helpers.hpp"
+#include "runtime/exec_env.h"
+#include "runtime/query_context.h"
+#include "runtime/runtime_state.h"
+#include "util/brpc_client_cache.h"
+#include "util/uid_util.h"
+
+namespace doris {
+
+RecCTESharedState::RecCTESharedState() = default;
+
+RecCTESharedState::~RecCTESharedState() = default;
+
+Status RecCTESharedState::emplace_block(RuntimeState* state, Block&& block) {
+ if (agg_data) {
+ auto num_rows = uint32_t(block.rows());
+ ColumnRawPtrs raw_columns;
+ std::vector<ColumnPtr> columns = block.get_columns_and_convert();
+ for (auto& col : columns) {
+ raw_columns.push_back(col.get());
+ }
+
+ std::visit(Overload {[&](std::monostate& arg) -> void {
+ throw
doris::Exception(ErrorCode::INTERNAL_ERROR,
+ "uninited hash table");
+ },
+ [&](auto& agg_method) -> void {
+ SCOPED_TIMER(hash_table_compute_timer);
+ using HashMethodType =
std::decay_t<decltype(agg_method)>;
+ using AggState = typename
HashMethodType::State;
+
+ AggState agg_state(raw_columns);
+ agg_method.init_serialized_keys(raw_columns,
num_rows);
+ distinct_row.clear();
+
+ size_t row = 0;
+ auto creator = [&](const auto& ctor, auto&
key, auto& origin) {
+ HashMethodType::try_presis_key(key,
origin, arena);
+ ctor(key);
+ distinct_row.push_back(row);
+ };
+ auto creator_for_null_key = [&]() {
distinct_row.push_back(row); };
+
+ SCOPED_TIMER(hash_table_emplace_timer);
+ lazy_emplace_batch_void(agg_method,
agg_state, num_rows, creator,
+ creator_for_null_key,
+ [&](uint32_t r) { row
= r; });
+ COUNTER_UPDATE(hash_table_input_counter,
num_rows);
+ }},
+ agg_data->method_variant);
+
+ if (distinct_row.size() == block.rows()) {
+ blocks.emplace_back(std::move(block));
+ } else if (!distinct_row.empty()) {
+ auto distinct_block = MutableBlock(block.clone_empty());
+ RETURN_IF_ERROR(block.append_to_block_by_selector(&distinct_block,
distinct_row));
+ blocks.emplace_back(distinct_block.to_block());
+ }
+ } else {
+ blocks.emplace_back(std::move(block));
+ }
+ return Status::OK();
+}
+
+PTransmitRecCTEBlockParams RecCTESharedState::build_basic_param(RuntimeState*
state,
+ const
TRecCTETarget& target) const {
+ PTransmitRecCTEBlockParams request;
+ request.set_node_id(target.node_id);
+
request.mutable_query_id()->CopyFrom(UniqueId(state->query_id()).to_proto());
+ request.mutable_fragment_instance_id()->CopyFrom(
+ UniqueId(target.fragment_instance_id).to_proto());
+ return request;
+}
+
+Status RecCTESharedState::send_data_to_targets(RuntimeState* state, size_t
round_offset) const {
+ if (targets.size() == 0) {
+ return Status::OK();
+ }
+ int send_multi_blocks_byte_size =
state->query_options().exchange_multi_blocks_byte_size;
+ int block_number_per_target =
+ int(blocks.size() - round_offset + targets.size() - 1) /
targets.size();
+ for (auto target : targets) {
+ auto stub =
state->get_query_ctx()->exec_env()->brpc_internal_client_cache()->get_client(
+ target.addr);
+ if (!stub) {
+ return Status::InternalError(fmt::format("Get rpc stub failed,
host={}, port={}",
+ target.addr.hostname,
target.addr.port));
+ }
+
+ // send blocks
+ int step = block_number_per_target;
+ while (round_offset < blocks.size() && step > 0) {
+ PTransmitRecCTEBlockParams request = build_basic_param(state,
target);
+ auto current_bytes = 0;
+ while (round_offset < blocks.size() && step > 0 &&
+ current_bytes < send_multi_blocks_byte_size) {
+ auto* pblock = request.add_blocks();
+ size_t uncompressed_bytes = 0;
+ size_t compressed_bytes = 0;
+ int64_t compress_time;
+ RETURN_IF_ERROR(blocks[round_offset].serialize(
+ state->be_exec_version(), pblock, &uncompressed_bytes,
&compressed_bytes,
+ &compress_time,
state->fragement_transmission_compression_type()));
+ round_offset++;
+ step--;
+ current_bytes += compressed_bytes;
+ }
+ request.set_eos(false);
+
+ PTransmitRecCTEBlockResult result;
+ brpc::Controller controller;
+ controller.set_timeout_ms(
+
get_execution_rpc_timeout_ms(state->get_query_ctx()->execution_timeout()));
+
+ stub->transmit_rec_cte_block(&controller, &request, &result,
brpc::DoNothing());
+ brpc::Join(controller.call_id());
+ RETURN_IF_ERROR(Status::create(result.status()));
+ }
+
+ // send eos
+ {
+ PTransmitRecCTEBlockParams request = build_basic_param(state,
target);
+ request.set_eos(true);
+
+ PTransmitRecCTEBlockResult result;
+ brpc::Controller controller;
+ stub->transmit_rec_cte_block(&controller, &request, &result,
brpc::DoNothing());
+ brpc::Join(controller.call_id());
+ RETURN_IF_ERROR(Status::create(result.status()));
+ }
+ }
+ return Status::OK();
+}
+
+} // namespace doris
diff --git a/be/src/exec/pipeline/rec_cte_shared_state.h
b/be/src/exec/pipeline/rec_cte_shared_state.h
index 527f98e1f0a..fe87f63e271 100644
--- a/be/src/exec/pipeline/rec_cte_shared_state.h
+++ b/be/src/exec/pipeline/rec_cte_shared_state.h
@@ -17,13 +17,18 @@
#pragma once
-#include "exec/common/distinct_agg_utils.h"
#include "exec/pipeline/dependency.h"
-#include "util/brpc_client_cache.h"
namespace doris {
+struct DistinctDataVariants;
+
struct RecCTESharedState : public BasicSharedState {
+ // Defined in rec_cte_shared_state.cpp: DistinctDataVariants carries the
+ // full hash-table variant machinery.
+ RecCTESharedState();
+ ~RecCTESharedState() override;
+
std::vector<TRecCTETarget> targets;
std::vector<Block> blocks;
IColumn::Selector distinct_row;
@@ -34,7 +39,11 @@ struct RecCTESharedState : public BasicSharedState {
RuntimeProfile::Counter* hash_table_emplace_timer = nullptr;
RuntimeProfile::Counter* hash_table_input_counter = nullptr;
- std::unique_ptr<DistinctDataVariants> agg_data = nullptr;
+ // No `= nullptr` initializer: a default member initializer makes GCC
+ // instantiate ~unique_ptr<DistinctDataVariants> in every TU that merely
sees
+ // this class, which needs the complete type. The defaulted constructor in
+ // rec_cte_shared_state.cpp already leaves this null.
+ std::unique_ptr<DistinctDataVariants> agg_data;
int current_round = 0;
int last_round_offset = 0;
@@ -47,130 +56,14 @@ struct RecCTESharedState : public BasicSharedState {
}
}
- Status emplace_block(RuntimeState* state, Block&& block) {
- if (agg_data) {
- auto num_rows = uint32_t(block.rows());
- ColumnRawPtrs raw_columns;
- std::vector<ColumnPtr> columns = block.get_columns_and_convert();
- for (auto& col : columns) {
- raw_columns.push_back(col.get());
- }
-
- std::visit(Overload {[&](std::monostate& arg) -> void {
- throw
doris::Exception(ErrorCode::INTERNAL_ERROR,
- "uninited hash
table");
- },
- [&](auto& agg_method) -> void {
- SCOPED_TIMER(hash_table_compute_timer);
- using HashMethodType =
std::decay_t<decltype(agg_method)>;
- using AggState = typename
HashMethodType::State;
-
- AggState agg_state(raw_columns);
-
agg_method.init_serialized_keys(raw_columns, num_rows);
- distinct_row.clear();
-
- size_t row = 0;
- auto creator = [&](const auto& ctor,
auto& key, auto& origin) {
- HashMethodType::try_presis_key(key,
origin, arena);
- ctor(key);
- distinct_row.push_back(row);
- };
- auto creator_for_null_key = [&]() {
- distinct_row.push_back(row);
- };
-
- SCOPED_TIMER(hash_table_emplace_timer);
- lazy_emplace_batch_void(agg_method,
agg_state, num_rows,
- creator,
creator_for_null_key,
- [&](uint32_t r) {
row = r; });
- COUNTER_UPDATE(hash_table_input_counter,
num_rows);
- }},
- agg_data->method_variant);
-
- if (distinct_row.size() == block.rows()) {
- blocks.emplace_back(std::move(block));
- } else if (!distinct_row.empty()) {
- auto distinct_block = MutableBlock(block.clone_empty());
-
RETURN_IF_ERROR(block.append_to_block_by_selector(&distinct_block,
distinct_row));
- blocks.emplace_back(distinct_block.to_block());
- }
- } else {
- blocks.emplace_back(std::move(block));
- }
- return Status::OK();
- }
+ // Bodies live in rec_cte_shared_state.cpp: they touch the distinct hash
+ // table variant machinery and the brpc client stack.
+ Status emplace_block(RuntimeState* state, Block&& block);
PTransmitRecCTEBlockParams build_basic_param(RuntimeState* state,
- const TRecCTETarget& target)
const {
- PTransmitRecCTEBlockParams request;
- request.set_node_id(target.node_id);
-
request.mutable_query_id()->CopyFrom(UniqueId(state->query_id()).to_proto());
- request.mutable_fragment_instance_id()->CopyFrom(
- UniqueId(target.fragment_instance_id).to_proto());
- return request;
- }
-
- Status send_data_to_targets(RuntimeState* state, size_t round_offset)
const {
- if (targets.size() == 0) {
- return Status::OK();
- }
- int send_multi_blocks_byte_size =
state->query_options().exchange_multi_blocks_byte_size;
- int block_number_per_target =
- int(blocks.size() - round_offset + targets.size() - 1) /
targets.size();
- for (auto target : targets) {
- auto stub =
-
state->get_query_ctx()->exec_env()->brpc_internal_client_cache()->get_client(
- target.addr);
- if (!stub) {
- return Status::InternalError(fmt::format("Get rpc stub failed,
host={}, port={}",
- target.addr.hostname,
target.addr.port));
- }
-
- // send blocks
- int step = block_number_per_target;
- while (round_offset < blocks.size() && step > 0) {
- PTransmitRecCTEBlockParams request = build_basic_param(state,
target);
- auto current_bytes = 0;
- while (round_offset < blocks.size() && step > 0 &&
- current_bytes < send_multi_blocks_byte_size) {
- auto* pblock = request.add_blocks();
- size_t uncompressed_bytes = 0;
- size_t compressed_bytes = 0;
- int64_t compress_time;
- RETURN_IF_ERROR(blocks[round_offset].serialize(
- state->be_exec_version(), pblock,
&uncompressed_bytes,
- &compressed_bytes, &compress_time,
- state->fragement_transmission_compression_type()));
- round_offset++;
- step--;
- current_bytes += compressed_bytes;
- }
- request.set_eos(false);
-
- PTransmitRecCTEBlockResult result;
- brpc::Controller controller;
- controller.set_timeout_ms(
-
get_execution_rpc_timeout_ms(state->get_query_ctx()->execution_timeout()));
+ const TRecCTETarget& target)
const;
- stub->transmit_rec_cte_block(&controller, &request, &result,
brpc::DoNothing());
- brpc::Join(controller.call_id());
- RETURN_IF_ERROR(Status::create(result.status()));
- }
-
- // send eos
- {
- PTransmitRecCTEBlockParams request = build_basic_param(state,
target);
- request.set_eos(true);
-
- PTransmitRecCTEBlockResult result;
- brpc::Controller controller;
- stub->transmit_rec_cte_block(&controller, &request, &result,
brpc::DoNothing());
- brpc::Join(controller.call_id());
- RETURN_IF_ERROR(Status::create(result.status()));
- }
- }
- return Status::OK();
- }
+ Status send_data_to_targets(RuntimeState* state, size_t round_offset)
const;
};
} // namespace doris
diff --git a/be/src/format/parquet/decoder.cpp
b/be/src/format/parquet/decoder.cpp
index f97ff464b80..016d69b4d12 100644
--- a/be/src/format/parquet/decoder.cpp
+++ b/be/src/format/parquet/decoder.cpp
@@ -20,6 +20,7 @@
#include <cctz/time_zone.h>
#include <gen_cpp/parquet_types.h>
+#include "common/cast_set.h"
#include "format/parquet/bool_plain_decoder.h"
#include "format/parquet/bool_rle_decoder.h"
#include "format/parquet/byte_array_dict_decoder.h"
@@ -28,8 +29,30 @@
#include "format/parquet/delta_bit_pack_decoder.h"
#include "format/parquet/fix_length_dict_decoder.hpp"
#include "format/parquet/fix_length_plain_decoder.h"
+#include "util/rle_encoding.h"
namespace doris {
+
+BaseDictDecoder::BaseDictDecoder() = default;
+
+BaseDictDecoder::~BaseDictDecoder() = default;
+
+Status BaseDictDecoder::set_data(Slice* data) {
+ _data = data;
+ _offset = 0;
+ uint8_t bit_width = *data->data;
+ _index_batch_decoder = std::make_unique<RleBatchDecoder<uint32_t>>(
+ reinterpret_cast<uint8_t*>(data->data) + 1,
static_cast<int>(data->size) - 1,
+ bit_width);
+ return Status::OK();
+}
+
+Status BaseDictDecoder::skip_values(size_t num_values) {
+ _indexes.resize(num_values);
+ _index_batch_decoder->GetBatch(_indexes.data(),
cast_set<uint32_t>(num_values));
+ return Status::OK();
+}
+
Status Decoder::get_decoder(tparquet::Type::type type,
tparquet::Encoding::type encoding,
std::unique_ptr<Decoder>& decoder) {
switch (encoding) {
diff --git a/be/src/format/parquet/decoder.h b/be/src/format/parquet/decoder.h
index c936e5e042f..cf1f3c81adf 100644
--- a/be/src/format/parquet/decoder.h
+++ b/be/src/format/parquet/decoder.h
@@ -38,13 +38,14 @@
#include "core/pod_array_fwd.h"
#include "core/types.h"
#include "format/parquet/parquet_common.h"
-#include "util/rle_encoding.h"
#include "util/slice.h"
namespace doris {
template <typename T>
class ColumnStr;
using ColumnString = ColumnStr<UInt32>;
+template <typename T>
+class RleBatchDecoder;
class Decoder {
public:
@@ -93,19 +94,16 @@ protected:
class BaseDictDecoder : public Decoder {
public:
- BaseDictDecoder() = default;
- ~BaseDictDecoder() override = default;
+ // Out-of-line: member unique_ptr<RleBatchDecoder<uint32_t>> only needs the
+ // complete type where the ctor/dtor/set_data/skip_values are defined
(decoder.cpp),
+ // keeping the costly RLE template machinery out of every includer of this
header.
+ // The ctor counts: a defaulted-in-class one would be defined in every TU
that
+ // constructs a derived decoder, and it odr-uses the member's destructor.
+ BaseDictDecoder();
+ ~BaseDictDecoder() override;
// Set the data to be decoded
- Status set_data(Slice* data) override {
- _data = data;
- _offset = 0;
- uint8_t bit_width = *data->data;
- _index_batch_decoder = std::make_unique<RleBatchDecoder<uint32_t>>(
- reinterpret_cast<uint8_t*>(data->data) + 1,
static_cast<int>(data->size) - 1,
- bit_width);
- return Status::OK();
- }
+ Status set_data(Slice* data) override;
protected:
/**
@@ -146,11 +144,7 @@ protected:
return Status::OK();
}
- Status skip_values(size_t num_values) override {
- _indexes.resize(num_values);
- _index_batch_decoder->GetBatch(_indexes.data(),
cast_set<uint32_t>(num_values));
- return Status::OK();
- }
+ Status skip_values(size_t num_values) override;
// For dictionary encoding
DorisUniqueBufferPtr<uint8_t> _dict;
diff --git a/be/src/format_v2/native/native_reader.h
b/be/src/format_v2/native/native_reader.h
index 15a52fe6f8b..8460a2faec5 100644
--- a/be/src/format_v2/native/native_reader.h
+++ b/be/src/format_v2/native/native_reader.h
@@ -25,6 +25,7 @@
#include <vector>
#include "format_v2/file_reader.h"
+#include "runtime/runtime_profile.h"
namespace doris::format::native {
diff --git a/be/src/format_v2/parquet/reader/native/decoder.cpp
b/be/src/format_v2/parquet/reader/native/decoder.cpp
index c068cafaf0d..845c9deacd7 100644
--- a/be/src/format_v2/parquet/reader/native/decoder.cpp
+++ b/be/src/format_v2/parquet/reader/native/decoder.cpp
@@ -20,6 +20,7 @@
#include <cctz/time_zone.h>
#include <gen_cpp/parquet_types.h>
+#include "common/cast_set.h"
#include "format_v2/parquet/reader/native/bool_plain_decoder.h"
#include "format_v2/parquet/reader/native/bool_rle_decoder.h"
#include "format_v2/parquet/reader/native/byte_array_dict_decoder.h"
@@ -28,8 +29,226 @@
#include "format_v2/parquet/reader/native/delta_bit_pack_decoder.h"
#include "format_v2/parquet/reader/native/fix_length_dict_decoder.hpp"
#include "format_v2/parquet/reader/native/fix_length_plain_decoder.h"
+#include "util/rle_encoding.h"
namespace doris::format::parquet::native {
+
+BaseDictDecoder::BaseDictDecoder() = default;
+
+BaseDictDecoder::~BaseDictDecoder() = default;
+
+Status BaseDictDecoder::set_data(Slice* data) {
+ if (UNLIKELY(data == nullptr || data->size == 0)) {
+ return Status::Corruption("Parquet dictionary index stream is empty");
+ }
+ _data = data;
+ _offset = 0;
+ uint8_t bit_width = *data->data;
+ // Dictionary indices are uint32_t; wider external widths make repeated
runs overwrite the
+ // decoder's four-byte state before any dictionary-bound check can run.
+ if (UNLIKELY(bit_width > 32)) {
+ return Status::Corruption("Parquet dictionary index bit width {}
exceeds 32", bit_width);
+ }
+ _index_batch_decoder = std::make_unique<RleBatchDecoder<uint32_t>>(
+ reinterpret_cast<uint8_t*>(data->data) + 1,
static_cast<int>(data->size) - 1,
+ bit_width);
+ return Status::OK();
+}
+
+Status BaseDictDecoder::decode_dictionary_indices(size_t num_values,
+ std::vector<uint32_t>*
indices) {
+ DORIS_CHECK(indices != nullptr);
+ indices->resize(num_values);
+ const auto decoded =
+ _index_batch_decoder->GetBatch(indices->data(),
cast_set<uint32_t>(num_values));
+ if (UNLIKELY(decoded != num_values)) {
+ return Status::IOError("Can't read enough Parquet dictionary indices");
+ }
+ const size_t num_dictionary_values = dictionary_size();
+ if (UNLIKELY(!dictionary_indices_in_bounds(indices->data(), num_values,
+ num_dictionary_values))) {
+ // The SIMD common path only computes a bound; recover the exact
corrupt row for the
+ // diagnostic after the batch has already been proven invalid.
+ for (size_t row = 0; row < num_values; ++row) {
+ if ((*indices)[row] < num_dictionary_values) {
+ continue;
+ }
+ return Status::Corruption(
+ "Parquet dictionary index {} at row {} exceeds dictionary
size {}",
+ (*indices)[row], row, num_dictionary_values);
+ }
+ }
+ return Status::OK();
+}
+
+Status BaseDictDecoder::decode_selected_dictionary_indices(const
ParquetSelection& selection,
+
std::vector<uint32_t>* indices) {
+ DORIS_CHECK(indices != nullptr);
+ const size_t num_dictionary_values = dictionary_size();
+ if (_is_fragmented_selection(selection)) {
+ RETURN_IF_ERROR(_decode_fragmented_selection(selection,
num_dictionary_values));
+ indices->assign(_skip_indices.begin(), _skip_indices.begin() +
selection.selected_values);
+ return Status::OK();
+ }
+ indices->resize(selection.selected_values);
+ size_t cursor = 0;
+ size_t output = 0;
+ for (const auto& range : selection.ranges) {
+ DORIS_CHECK(range.first >= cursor);
+ RETURN_IF_ERROR(
+ _decode_and_validate_skipped(range.first - cursor, cursor,
num_dictionary_values));
+ const auto decoded = _index_batch_decoder->GetBatch(indices->data() +
output,
+
cast_set<uint32_t>(range.count));
+ if (UNLIKELY(decoded != range.count)) {
+ return Status::IOError("Can't read enough Parquet dictionary
indices");
+ }
+ if (UNLIKELY(!dictionary_indices_in_bounds(indices->data() + output,
range.count,
+ num_dictionary_values))) {
+ for (size_t row = 0; row < range.count; ++row) {
+ if ((*indices)[output + row] < num_dictionary_values) {
+ continue;
+ }
+ return Status::Corruption(
+ "Parquet dictionary index {} at row {} exceeds
dictionary size {}",
+ (*indices)[output + row], range.first + row,
num_dictionary_values);
+ }
+ }
+ output += range.count;
+ cursor = range.first + range.count;
+ }
+ DORIS_CHECK(cursor <= selection.total_values);
+ RETURN_IF_ERROR(_decode_and_validate_skipped(selection.total_values -
cursor, cursor,
+ num_dictionary_values));
+ DORIS_CHECK_EQ(output, selection.selected_values);
+ return Status::OK();
+}
+
+Status BaseDictDecoder::_decode_fragmented_selection(const ParquetSelection&
selection,
+ size_t
num_dictionary_values) {
+ // Decode and validate the page batch once when predicate survivors
alternate in tiny runs.
+ // Walking each range separately turns one RLE batch into millions of
decoder calls for
+ // low-cardinality predicates such as TPC-DS quantity buckets.
+ _skip_indices.resize(selection.total_values);
+ const auto decoded = _index_batch_decoder->GetBatch(_skip_indices.data(),
+
cast_set<uint32_t>(selection.total_values));
+ if (UNLIKELY(decoded != selection.total_values)) {
+ return Status::IOError("Can't read enough Parquet dictionary indices");
+ }
+ if (UNLIKELY(!dictionary_indices_in_bounds(_skip_indices.data(),
selection.total_values,
+ num_dictionary_values))) {
+ for (size_t row = 0; row < selection.total_values; ++row) {
+ if (_skip_indices[row] < num_dictionary_values) {
+ continue;
+ }
+ return Status::Corruption(
+ "Parquet dictionary index {} at row {} exceeds dictionary
size {}",
+ _skip_indices[row], row, num_dictionary_values);
+ }
+ }
+ size_t output = 0;
+ constexpr size_t MAX_INLINE_COPY_VALUES = 4;
+ for (const auto& range : selection.ranges) {
+ DORIS_CHECK(range.first + range.count <= selection.total_values);
+ // Alternating predicates mostly produce one-row spans; inline tiny
forward copies so
+ // range compaction does not replace decoder calls with equally
numerous libc calls.
+ if (range.count <= MAX_INLINE_COPY_VALUES) {
+ for (size_t row = 0; row < range.count; ++row) {
+ _skip_indices[output + row] = _skip_indices[range.first + row];
+ }
+ } else {
+ memmove(_skip_indices.data() + output, _skip_indices.data() +
range.first,
+ range.count * sizeof(uint32_t));
+ }
+ output += range.count;
+ }
+ DORIS_CHECK_EQ(output, selection.selected_values);
+ return Status::OK();
+}
+
+Status BaseDictDecoder::_decode_and_validate_skipped(size_t num_values, size_t
row_offset,
+ size_t
num_dictionary_values) {
+ constexpr size_t kSkipBatchSize = 4096;
+ // Skipped dictionary ids are still external input and must be
bounds-checked, but keeping
+ // only one bounded gap buffer avoids the page-sized scratch used by
sparse selections.
+ _skip_indices.resize(std::min(num_values, kSkipBatchSize));
+ size_t skipped_values = 0;
+ while (skipped_values < num_values) {
+ const size_t batch_size = std::min(num_values - skipped_values,
kSkipBatchSize);
+ const auto skipped =
_index_batch_decoder->GetBatch(_skip_indices.data(),
+
static_cast<uint32_t>(batch_size));
+ if (UNLIKELY(skipped != batch_size)) {
+ return Status::IOError(
+ "Can't skip enough Parquet dictionary indices at row {}:
{} of {}",
+ row_offset + skipped_values, skipped, batch_size);
+ }
+ // Filter gaps may be huge RLE runs; validate them in bounded
SIMD-sized batches.
+ if (UNLIKELY(!dictionary_indices_in_bounds(_skip_indices.data(),
batch_size,
+ num_dictionary_values))) {
+ for (size_t row = 0; row < batch_size; ++row) {
+ if (_skip_indices[row] < num_dictionary_values) {
+ continue;
+ }
+ return Status::Corruption(
+ "Parquet dictionary index {} at skipped row {} exceeds
dictionary "
+ "size {}",
+ _skip_indices[row], row_offset + skipped_values + row,
+ num_dictionary_values);
+ }
+ }
+ skipped_values += batch_size;
+ }
+ return Status::OK();
+}
+
+Status BaseDictDecoder::_decode_dictionary_values(size_t num_values, size_t
row_offset,
+ size_t num_dictionary_values,
+
ParquetDictionaryValueConsumer& consumer) {
+ constexpr size_t kLiteralBatchSize = 1024;
+ size_t decoded_values = 0;
+ while (decoded_values < num_values) {
+ const int32_t repeats = _index_batch_decoder->NextNumRepeats();
+ if (repeats > 0) {
+ const size_t run = std::min<size_t>(repeats, num_values -
decoded_values);
+ const uint32_t index =
_index_batch_decoder->GetRepeatedValue(cast_set<int32_t>(run));
+ if (UNLIKELY(static_cast<size_t>(index) >= num_dictionary_values))
{
+ return Status::Corruption(
+ "Parquet dictionary index {} at row {} exceeds
dictionary size {}", index,
+ row_offset + decoded_values, num_dictionary_values);
+ }
+ RETURN_IF_ERROR(consumer.consume_repeated(index, run));
+ decoded_values += run;
+ continue;
+ }
+
+ const int32_t literals = _index_batch_decoder->NextNumLiterals();
+ if (UNLIKELY(literals == 0)) {
+ return Status::IOError("Can't read enough Parquet dictionary
indices");
+ }
+ const size_t batch = std::min(
+ {static_cast<size_t>(literals), num_values - decoded_values,
kLiteralBatchSize});
+ _skip_indices.resize(batch);
+ if
(UNLIKELY(!_index_batch_decoder->GetLiteralValues(cast_set<int32_t>(batch),
+
_skip_indices.data()))) {
+ return Status::IOError("Can't read enough Parquet dictionary
indices");
+ }
+ if (UNLIKELY(!dictionary_indices_in_bounds(_skip_indices.data(), batch,
+ num_dictionary_values))) {
+ for (size_t row = 0; row < batch; ++row) {
+ if (_skip_indices[row] < num_dictionary_values) {
+ continue;
+ }
+ return Status::Corruption(
+ "Parquet dictionary index {} at row {} exceeds
dictionary size {}",
+ _skip_indices[row], row_offset + decoded_values + row,
+ num_dictionary_values);
+ }
+ }
+ RETURN_IF_ERROR(consumer.consume_indices(_skip_indices.data(), batch));
+ decoded_values += batch;
+ }
+ return Status::OK();
+}
+
namespace {
Status unsupported_type(tparquet::Type::type type, tparquet::Encoding::type
encoding) {
return Status::InternalError("Unsupported type {}(encoding={}) in parquet
decoder",
diff --git a/be/src/format_v2/parquet/reader/native/decoder.h
b/be/src/format_v2/parquet/reader/native/decoder.h
index 628a34ac976..873a7ab9b02 100644
--- a/be/src/format_v2/parquet/reader/native/decoder.h
+++ b/be/src/format_v2/parquet/reader/native/decoder.h
@@ -37,9 +37,13 @@
#include "core/custom_allocator.h"
#include "core/data_type_serde/parquet_decode_source.h"
#include "core/types.h"
-#include "util/rle_encoding.h"
#include "util/slice.h"
+namespace doris {
+template <typename T>
+class RleBatchDecoder;
+} // namespace doris
+
namespace doris::format::parquet::native {
inline bool dictionary_indices_in_bounds(const uint32_t* indices, size_t count,
@@ -127,99 +131,25 @@ protected:
class BaseDictDecoder : public Decoder {
public:
- BaseDictDecoder() = default;
- ~BaseDictDecoder() override = default;
+ // Out-of-line together with every member that dereferences
_index_batch_decoder:
+ // keeping the RleBatchDecoder<uint32_t> machinery out of this header
stops every
+ // includer from instantiating the RLE decode chain; the complete type
lives only
+ // in decoder.cpp. All of these are per-page/per-batch virtual calls. The
ctor is
+ // out of line for the same reason: a defaulted-in-class one would be
defined in
+ // every TU that constructs a derived decoder, and it odr-uses the
member's dtor.
+ BaseDictDecoder();
+ ~BaseDictDecoder() override;
// Set the data to be decoded
- Status set_data(Slice* data) override {
- if (UNLIKELY(data == nullptr || data->size == 0)) {
- return Status::Corruption("Parquet dictionary index stream is
empty");
- }
- _data = data;
- _offset = 0;
- uint8_t bit_width = *data->data;
- // Dictionary indices are uint32_t; wider external widths make
repeated runs overwrite the
- // decoder's four-byte state before any dictionary-bound check can run.
- if (UNLIKELY(bit_width > 32)) {
- return Status::Corruption("Parquet dictionary index bit width {}
exceeds 32",
- bit_width);
- }
- _index_batch_decoder = std::make_unique<RleBatchDecoder<uint32_t>>(
- reinterpret_cast<uint8_t*>(data->data) + 1,
static_cast<int>(data->size) - 1,
- bit_width);
- return Status::OK();
- }
+ Status set_data(Slice* data) override;
bool has_dictionary() const override { return true; }
uint64_t dictionary_generation() const override { return
_dictionary_generation; }
- Status decode_dictionary_indices(size_t num_values, std::vector<uint32_t>*
indices) override {
- DORIS_CHECK(indices != nullptr);
- indices->resize(num_values);
- const auto decoded =
- _index_batch_decoder->GetBatch(indices->data(),
cast_set<uint32_t>(num_values));
- if (UNLIKELY(decoded != num_values)) {
- return Status::IOError("Can't read enough Parquet dictionary
indices");
- }
- const size_t num_dictionary_values = dictionary_size();
- if (UNLIKELY(!dictionary_indices_in_bounds(indices->data(), num_values,
- num_dictionary_values))) {
- // The SIMD common path only computes a bound; recover the exact
corrupt row for the
- // diagnostic after the batch has already been proven invalid.
- for (size_t row = 0; row < num_values; ++row) {
- if ((*indices)[row] < num_dictionary_values) {
- continue;
- }
- return Status::Corruption(
- "Parquet dictionary index {} at row {} exceeds
dictionary size {}",
- (*indices)[row], row, num_dictionary_values);
- }
- }
- return Status::OK();
- }
+ Status decode_dictionary_indices(size_t num_values, std::vector<uint32_t>*
indices) override;
Status decode_selected_dictionary_indices(const ParquetSelection&
selection,
- std::vector<uint32_t>* indices)
override {
- DORIS_CHECK(indices != nullptr);
- const size_t num_dictionary_values = dictionary_size();
- if (_is_fragmented_selection(selection)) {
- RETURN_IF_ERROR(_decode_fragmented_selection(selection,
num_dictionary_values));
- indices->assign(_skip_indices.begin(),
- _skip_indices.begin() + selection.selected_values);
- return Status::OK();
- }
- indices->resize(selection.selected_values);
- size_t cursor = 0;
- size_t output = 0;
- for (const auto& range : selection.ranges) {
- DORIS_CHECK(range.first >= cursor);
- RETURN_IF_ERROR(_decode_and_validate_skipped(range.first - cursor,
cursor,
-
num_dictionary_values));
- const auto decoded =
_index_batch_decoder->GetBatch(indices->data() + output,
-
cast_set<uint32_t>(range.count));
- if (UNLIKELY(decoded != range.count)) {
- return Status::IOError("Can't read enough Parquet dictionary
indices");
- }
- if (UNLIKELY(!dictionary_indices_in_bounds(indices->data() +
output, range.count,
-
num_dictionary_values))) {
- for (size_t row = 0; row < range.count; ++row) {
- if ((*indices)[output + row] < num_dictionary_values) {
- continue;
- }
- return Status::Corruption(
- "Parquet dictionary index {} at row {} exceeds
dictionary size {}",
- (*indices)[output + row], range.first + row,
num_dictionary_values);
- }
- }
- output += range.count;
- cursor = range.first + range.count;
- }
- DORIS_CHECK(cursor <= selection.total_values);
- RETURN_IF_ERROR(_decode_and_validate_skipped(selection.total_values -
cursor, cursor,
- num_dictionary_values));
- DORIS_CHECK_EQ(output, selection.selected_values);
- return Status::OK();
- }
+ std::vector<uint32_t>* indices)
override;
Status decode_dictionary_values(size_t num_values,
ParquetDictionaryValueConsumer& consumer)
override {
@@ -266,135 +196,18 @@ protected:
}
Status _decode_fragmented_selection(const ParquetSelection& selection,
- size_t num_dictionary_values) {
- // Decode and validate the page batch once when predicate survivors
alternate in tiny runs.
- // Walking each range separately turns one RLE batch into millions of
decoder calls for
- // low-cardinality predicates such as TPC-DS quantity buckets.
- _skip_indices.resize(selection.total_values);
- const auto decoded = _index_batch_decoder->GetBatch(
- _skip_indices.data(),
cast_set<uint32_t>(selection.total_values));
- if (UNLIKELY(decoded != selection.total_values)) {
- return Status::IOError("Can't read enough Parquet dictionary
indices");
- }
- if (UNLIKELY(!dictionary_indices_in_bounds(_skip_indices.data(),
selection.total_values,
- num_dictionary_values))) {
- for (size_t row = 0; row < selection.total_values; ++row) {
- if (_skip_indices[row] < num_dictionary_values) {
- continue;
- }
- return Status::Corruption(
- "Parquet dictionary index {} at row {} exceeds
dictionary size {}",
- _skip_indices[row], row, num_dictionary_values);
- }
- }
- size_t output = 0;
- constexpr size_t MAX_INLINE_COPY_VALUES = 4;
- for (const auto& range : selection.ranges) {
- DORIS_CHECK(range.first + range.count <= selection.total_values);
- // Alternating predicates mostly produce one-row spans; inline
tiny forward copies so
- // range compaction does not replace decoder calls with equally
numerous libc calls.
- if (range.count <= MAX_INLINE_COPY_VALUES) {
- for (size_t row = 0; row < range.count; ++row) {
- _skip_indices[output + row] = _skip_indices[range.first +
row];
- }
- } else {
- memmove(_skip_indices.data() + output, _skip_indices.data() +
range.first,
- range.count * sizeof(uint32_t));
- }
- output += range.count;
- }
- DORIS_CHECK_EQ(output, selection.selected_values);
- return Status::OK();
- }
+ size_t num_dictionary_values);
Status skip_values(size_t num_values) override {
return _decode_and_validate_skipped(num_values, 0, dictionary_size());
}
Status _decode_and_validate_skipped(size_t num_values, size_t row_offset,
- size_t num_dictionary_values) {
- constexpr size_t kSkipBatchSize = 4096;
- // Skipped dictionary ids are still external input and must be
bounds-checked, but keeping
- // only one bounded gap buffer avoids the page-sized scratch used by
sparse selections.
- _skip_indices.resize(std::min(num_values, kSkipBatchSize));
- size_t skipped_values = 0;
- while (skipped_values < num_values) {
- const size_t batch_size = std::min(num_values - skipped_values,
kSkipBatchSize);
- const auto skipped =
_index_batch_decoder->GetBatch(_skip_indices.data(),
-
static_cast<uint32_t>(batch_size));
- if (UNLIKELY(skipped != batch_size)) {
- return Status::IOError(
- "Can't skip enough Parquet dictionary indices at row
{}: {} of {}",
- row_offset + skipped_values, skipped, batch_size);
- }
- // Filter gaps may be huge RLE runs; validate them in bounded
SIMD-sized batches.
- if (UNLIKELY(!dictionary_indices_in_bounds(_skip_indices.data(),
batch_size,
-
num_dictionary_values))) {
- for (size_t row = 0; row < batch_size; ++row) {
- if (_skip_indices[row] < num_dictionary_values) {
- continue;
- }
- return Status::Corruption(
- "Parquet dictionary index {} at skipped row {}
exceeds dictionary "
- "size {}",
- _skip_indices[row], row_offset + skipped_values +
row,
- num_dictionary_values);
- }
- }
- skipped_values += batch_size;
- }
- return Status::OK();
- }
+ size_t num_dictionary_values);
Status _decode_dictionary_values(size_t num_values, size_t row_offset,
size_t num_dictionary_values,
- ParquetDictionaryValueConsumer& consumer)
{
- constexpr size_t kLiteralBatchSize = 1024;
- size_t decoded_values = 0;
- while (decoded_values < num_values) {
- const int32_t repeats = _index_batch_decoder->NextNumRepeats();
- if (repeats > 0) {
- const size_t run = std::min<size_t>(repeats, num_values -
decoded_values);
- const uint32_t index =
-
_index_batch_decoder->GetRepeatedValue(cast_set<int32_t>(run));
- if (UNLIKELY(static_cast<size_t>(index) >=
num_dictionary_values)) {
- return Status::Corruption(
- "Parquet dictionary index {} at row {} exceeds
dictionary size {}",
- index, row_offset + decoded_values,
num_dictionary_values);
- }
- RETURN_IF_ERROR(consumer.consume_repeated(index, run));
- decoded_values += run;
- continue;
- }
-
- const int32_t literals = _index_batch_decoder->NextNumLiterals();
- if (UNLIKELY(literals == 0)) {
- return Status::IOError("Can't read enough Parquet dictionary
indices");
- }
- const size_t batch = std::min({static_cast<size_t>(literals),
- num_values - decoded_values,
kLiteralBatchSize});
- _skip_indices.resize(batch);
- if
(UNLIKELY(!_index_batch_decoder->GetLiteralValues(cast_set<int32_t>(batch),
-
_skip_indices.data()))) {
- return Status::IOError("Can't read enough Parquet dictionary
indices");
- }
- if (UNLIKELY(!dictionary_indices_in_bounds(_skip_indices.data(),
batch,
-
num_dictionary_values))) {
- for (size_t row = 0; row < batch; ++row) {
- if (_skip_indices[row] < num_dictionary_values) {
- continue;
- }
- return Status::Corruption(
- "Parquet dictionary index {} at row {} exceeds
dictionary size {}",
- _skip_indices[row], row_offset + decoded_values +
row,
- num_dictionary_values);
- }
- }
- RETURN_IF_ERROR(consumer.consume_indices(_skip_indices.data(),
batch));
- decoded_values += batch;
- }
- return Status::OK();
- }
+ ParquetDictionaryValueConsumer& consumer);
// For dictionary encoding
DorisUniqueBufferPtr<uint8_t> _dict;
diff --git a/be/src/storage/compaction/compaction.cpp
b/be/src/storage/compaction/compaction.cpp
index 46dc32d2bcd..86578831740 100644
--- a/be/src/storage/compaction/compaction.cpp
+++ b/be/src/storage/compaction/compaction.cpp
@@ -32,6 +32,7 @@
#include <nlohmann/json.hpp>
#include <numeric>
#include <ostream>
+#include <ranges>
#include <set>
#include <shared_mutex>
#include <utility>
diff --git a/be/src/storage/index/ann/ann_index_writer.h
b/be/src/storage/index/ann/ann_index_writer.h
index 67061bef921..913a4e33084 100644
--- a/be/src/storage/index/ann/ann_index_writer.h
+++ b/be/src/storage/index/ann/ann_index_writer.h
@@ -17,9 +17,20 @@
#pragma once
+// CLucene is third-party code and is not clean under -Wconversion (which
+// -Wshorten-64-to-32 belongs to). Whether its first expansion lands inside
+// someone else's suppressed region depends on include order, so suppress it
+// deliberately here (same pattern as inverted_index_common_impl.h).
+#ifdef __clang__
+#pragma clang diagnostic push
+#pragma clang diagnostic ignored "-Wconversion"
+#endif
#include <CLucene.h> // IWYU pragma: keep
#include <CLucene/analysis/LanguageBasedAnalyzer.h>
#include <CLucene/util/bkd/bkd_writer.h>
+#ifdef __clang__
+#pragma clang diagnostic pop
+#endif
#include <glog/logging.h>
#include <cstdint>
diff --git a/be/src/storage/index/index_file_writer.h
b/be/src/storage/index/index_file_writer.h
index a303de8b68c..35e8904b837 100644
--- a/be/src/storage/index/index_file_writer.h
+++ b/be/src/storage/index/index_file_writer.h
@@ -17,8 +17,19 @@
#pragma once
+// CLucene is third-party code and is not clean under -Wconversion (which
+// -Wshorten-64-to-32 belongs to). Whether its first expansion lands inside
+// someone else's suppressed region depends on include order, so suppress it
+// deliberately here (same pattern as inverted_index_common_impl.h).
+#ifdef __clang__
+#pragma clang diagnostic push
+#pragma clang diagnostic ignored "-Wconversion"
+#endif
#include <CLucene.h> // IWYU pragma: keep
#include <CLucene/store/IndexInput.h>
+#ifdef __clang__
+#pragma clang diagnostic pop
+#endif
#include <gen_cpp/olap_common.pb.h>
#include <gen_cpp/olap_file.pb.h>
diff --git a/be/src/storage/index/inverted/inverted_index_searcher.h
b/be/src/storage/index/inverted/inverted_index_searcher.h
index 63009a6e2ee..97e3c2dd189 100644
--- a/be/src/storage/index/inverted/inverted_index_searcher.h
+++ b/be/src/storage/index/inverted/inverted_index_searcher.h
@@ -21,10 +21,21 @@
#pragma clang diagnostic push
#pragma clang diagnostic ignored "-Wshadow-field"
#endif
+// CLucene is third-party code and is not clean under -Wconversion (which
+// -Wshorten-64-to-32 belongs to). Whether its first expansion lands inside
+// someone else's suppressed region depends on include order, so suppress it
+// deliberately here (same pattern as inverted_index_common_impl.h).
+#ifdef __clang__
+#pragma clang diagnostic push
+#pragma clang diagnostic ignored "-Wconversion"
+#endif
#include <CLucene.h> // IWYU pragma: keep
#ifdef __clang__
#pragma clang diagnostic pop
#endif
+#ifdef __clang__
+#pragma clang diagnostic pop
+#endif
#include <memory>
#include <optional>
diff --git a/be/src/storage/index/inverted/query/query.h
b/be/src/storage/index/inverted/query/query.h
index 4f985736a3b..a64d9034f55 100644
--- a/be/src/storage/index/inverted/query/query.h
+++ b/be/src/storage/index/inverted/query/query.h
@@ -17,9 +17,20 @@
#pragma once
+// CLucene is third-party code and is not clean under -Wconversion (which
+// -Wshorten-64-to-32 belongs to). Whether its first expansion lands inside
+// someone else's suppressed region depends on include order, so suppress it
+// deliberately here (same pattern as inverted_index_common_impl.h).
+#ifdef __clang__
+#pragma clang diagnostic push
+#pragma clang diagnostic ignored "-Wconversion"
+#endif
#include <CLucene.h> // IWYU pragma: keep
#include <CLucene/index/IndexReader.h>
#include <CLucene/index/Term.h>
+#ifdef __clang__
+#pragma clang diagnostic pop
+#endif
#include <gen_cpp/PaloInternalService_types.h>
#include <memory>
diff --git a/be/src/storage/index/inverted/similarity/bm25_similarity.cpp
b/be/src/storage/index/inverted/similarity/bm25_similarity.cpp
index b973243c489..2ab946a8791 100644
--- a/be/src/storage/index/inverted/similarity/bm25_similarity.cpp
+++ b/be/src/storage/index/inverted/similarity/bm25_similarity.cpp
@@ -21,8 +21,6 @@
namespace doris::segment_v2 {
-using namespace inverted_index;
-
const int32_t BM25Similarity::MAX_INT32 = std::numeric_limits<int32_t>::max();
const uint32_t BM25Similarity::MAX_INT4 =
long_to_int4(static_cast<uint64_t>(MAX_INT32));
const int32_t BM25Similarity::NUM_FREE_VALUES = 255 -
static_cast<int>(MAX_INT4);
diff --git a/be/src/storage/olap_define.h b/be/src/storage/olap_define.h
index 5131c51ca01..2f40e8d6ab1 100644
--- a/be/src/storage/olap_define.h
+++ b/be/src/storage/olap_define.h
@@ -157,11 +157,14 @@ const std::string ROWSET_ID_PREFIX = "s_";
const std::string REMOTE_ROWSET_GC_PREFIX = "gc_";
const std::string REMOTE_TABLET_GC_PREFIX = "tgc_";
-// Declare copy constructor and equal operator as private
+// Declare copy constructor and equal operator as private.
+// No trailing semicolon: call sites write `DISALLOW_COPY_AND_ASSIGN(X);`, same
+// contract as butil/macros.h's definition of the same macro (which wins in TUs
+// that include butil first — keep the two expansions call-site compatible).
#ifndef DISALLOW_COPY_AND_ASSIGN
#define DISALLOW_COPY_AND_ASSIGN(type_t) \
type_t& operator=(const type_t&); \
- type_t(const type_t&);
+ type_t(const type_t&)
#endif
#define SAFE_DELETE(ptr) \
diff --git a/be/src/storage/tablet/base_tablet.cpp
b/be/src/storage/tablet/base_tablet.cpp
index aab47367c7d..ec16d69bf57 100644
--- a/be/src/storage/tablet/base_tablet.cpp
+++ b/be/src/storage/tablet/base_tablet.cpp
@@ -26,6 +26,7 @@
#include <cstdint>
#include <iterator>
#include <random>
+#include <ranges>
#include <shared_mutex>
#include "cloud/cloud_tablet.h"
diff --git a/be/test/core/block/block_test.cpp
b/be/test/core/block/block_test.cpp
index 2a3f07e01c5..566eaccc982 100644
--- a/be/test/core/block/block_test.cpp
+++ b/be/test/core/block/block_test.cpp
@@ -65,6 +65,7 @@
#include "exec/common/sip_hash.h"
#include "runtime/descriptor_helper.h"
#include "runtime/descriptors.h"
+#include "runtime/runtime_profile.h"
#include "testutil/column_helper.h"
#include "util/debug_points.h"
#include "util/defer_op.h"
diff --git a/be/test/core/column/column_variant_v2_test.cpp
b/be/test/core/column/column_variant_v2_test.cpp
index efee3303224..91656d4b526 100644
--- a/be/test/core/column/column_variant_v2_test.cpp
+++ b/be/test/core/column/column_variant_v2_test.cpp
@@ -61,6 +61,7 @@
#include "exec/common/hash_table/hash_map_context.h"
#include "exec/common/hash_table/string_hash_map.h"
#include "exec/common/sip_hash.h"
+#include "exec/sort/hybrid_sorter.h"
#include "exprs/function/parse/variant_jsonb_parse.h"
#include "exprs/function/parse/variant_string_parse.h"
#include "runtime/memory/mem_tracker.h"
diff --git a/be/test/exec/operator/asof_join_test.cpp
b/be/test/exec/operator/asof_join_test.cpp
index 8b31897b7de..accf929fd5c 100644
--- a/be/test/exec/operator/asof_join_test.cpp
+++ b/be/test/exec/operator/asof_join_test.cpp
@@ -16,6 +16,13 @@
// under the License.
#include <gtest/gtest.h>
+// Must precede exec/common/join_utils.h below.
AsofIndexGroup::sort_and_finalize calls the
+// global pdqsort, which ADL cannot reach from std::vector's iterators, so the
name has to be
+// visible where the template is defined rather than where it is instantiated.
join_utils.h
+// deliberately does not include it: core/column/column.h stopped pulling
exec/sort/hybrid_sorter.h
+// (pdqsort + timsort) out of 808 TUs, and production TUs reach join_utils.h
through headers that
+// already supply pdqsort. Only this test pays for it.
+#include <pdqsort.h>
#include <initializer_list>
#include <type_traits>
diff --git a/be/test/format/parquet/fix_length_dict_decoder_empty_dict_test.cpp
b/be/test/format/parquet/fix_length_dict_decoder_empty_dict_test.cpp
index e901507b0c3..4eb7180d70e 100644
--- a/be/test/format/parquet/fix_length_dict_decoder_empty_dict_test.cpp
+++ b/be/test/format/parquet/fix_length_dict_decoder_empty_dict_test.cpp
@@ -18,7 +18,16 @@
#include <gtest/gtest.h>
#include "core/column/column_vector.h"
+// Must precede the header below. FixLengthDictDecoder::_decode_values
dereferences the inherited
+// unique_ptr<RleBatchDecoder<uint32_t>>, and that type depends on no template
parameter, so the
+// call is bound when the template is parsed rather than when it is
instantiated. decoder.h only
+// forward-declares RleBatchDecoder to keep the RLE machinery out of its ~530
includers, so this TU
+// supplies the complete type up front. Only this test pays for it.
+// clang-format off
+#include "util/rle_encoding.h" // IWYU pragma: keep
+
#include "format/parquet/fix_length_dict_decoder.hpp"
+// clang-format on
namespace doris {
diff --git a/be/test/format/parquet/fix_length_dict_decoder_test.cpp
b/be/test/format/parquet/fix_length_dict_decoder_test.cpp
index 5c8854b665b..22015555abd 100644
--- a/be/test/format/parquet/fix_length_dict_decoder_test.cpp
+++ b/be/test/format/parquet/fix_length_dict_decoder_test.cpp
@@ -15,7 +15,16 @@
// specific language governing permissions and limitations
// under the License.
+// Must precede the header under test. FixLengthDictDecoder::_decode_values
dereferences the
+// inherited unique_ptr<RleBatchDecoder<uint32_t>>, and that type depends on
no template parameter,
+// so the call is bound when the template is parsed rather than when it is
instantiated. decoder.h
+// only forward-declares RleBatchDecoder to keep the RLE machinery out of its
~530 includers, so
+// this TU supplies the complete type up front. Only this test pays for it.
+// clang-format off
+#include "util/rle_encoding.h" // IWYU pragma: keep
+
#include "format/parquet/fix_length_dict_decoder.hpp"
+// clang-format on
#include <gtest/gtest.h>
diff --git a/be/test/storage/segment/hierarchical_data_iterator_test.cpp
b/be/test/storage/segment/hierarchical_data_iterator_test.cpp
index 69619726e4f..b69789d02c5 100644
--- a/be/test/storage/segment/hierarchical_data_iterator_test.cpp
+++ b/be/test/storage/segment/hierarchical_data_iterator_test.cpp
@@ -190,7 +190,7 @@ public:
return Status::InvalidArgument("JSONB destination is not a string
column");
}
- const size_t produced = std::min(*rows, ROWS -
_state->current_ordinal);
+ const size_t produced = std::min<size_t>(*rows, ROWS -
_state->current_ordinal);
auto serde = std::make_shared<doris::DataTypeJsonb>()->get_serde();
doris::DataTypeSerDe::FormatOptions options;
for (size_t row = 0; row < produced; ++row) {
@@ -241,7 +241,7 @@ public:
return Status::InvalidArgument("sparse destination is not a map");
}
- const size_t produced = std::min(*rows, ROWS -
_state->current_ordinal);
+ const size_t produced = std::min<size_t>(*rows, ROWS -
_state->current_ordinal);
auto& keys = assert_cast<ColumnString&>(map->get_keys());
auto& values = assert_cast<ColumnString&>(map->get_values());
auto& map_offsets = map->get_offsets();
@@ -297,7 +297,7 @@ public:
return Status::InvalidArgument("JSONB sparse destination is not a
map");
}
- const size_t produced = std::min(*rows, ROWS - _current_ordinal);
+ const size_t produced = std::min<size_t>(*rows, ROWS -
_current_ordinal);
auto& keys = assert_cast<ColumnString&>(map->get_keys());
auto& values = assert_cast<ColumnString&>(map->get_values());
auto& offsets = map->get_offsets();
diff --git a/build-support/check-header-deps.py
b/build-support/check-header-deps.py
index c91a3f25b4e..7161b2e15fd 100755
--- a/build-support/check-header-deps.py
+++ b/build-support/check-header-deps.py
@@ -197,8 +197,110 @@ RULES = [
"thrift and forward-declared); this was a dead include spreading
PBlock "
"and segment_v2.pb.h to ~595 TUs through thread_context.h",
),
+ (
+ "core/pod_array.h",
+ "runtime/thread_context.h",
+ set(),
+ "dead include left over from the PODArray memory-tracking experiment "
+ "(#50549); the tracking logic since moved into Allocator and
pod_array.h "
+ "references no thread_context symbol, yet the edge dragged
thread_context, "
+ "exec_env.h and mem_tracker_limiter.h into 203 TUs of core/",
+ ),
+ (
+ "core/column/column.h",
+ "exec/sort/",
+ set(),
+ "core must not depend on the exec sort machinery: column.h only names "
+ "HybridSorter in virtual signatures (forward-declared, bodies in "
+ "column.cpp); the old hybrid_sorter.h include was a layering violation
"
+ "that pushed pdqsort/timsort into 808 TUs "
+ "(exec/common/endian.h still rides in via storage/olap_common.h -> "
+ "util/hash_util.hpp, a separate pre-existing wart)",
+ ),
+ (
+ "format/parquet/decoder.h",
+ "util/rle_encoding.h",
+ set(),
+ "BaseDictDecoder holds RleBatchDecoder<uint32_t> behind a unique_ptr "
+ "(forward-declared; the dtor and every member that dereferences it are
"
+ "defined in decoder.cpp); the old include made every one of ~530 TUs "
+ "that transitively see a parquet decoder instantiate the whole "
+ "RLE/BitPacking decode chain at ~0.4 CPU s each",
+ ),
+ (
+ "format_v2/parquet/reader/native/decoder.h",
+ "util/rle_encoding.h",
+ set(),
+ "same contract as format/parquet/decoder.h: the dictionary index "
+ "decoder is forward-declared and only decoder.cpp needs the complete "
+ "RleBatchDecoder type; keeping rle_encoding.h (and the unrolled "
+ "bit_packing.inline.h it carries) out of this header keeps the RLE "
+ "instantiation chain out of the native-reader include tree",
+ ),
+ (
+ "exec/pipeline/dependency.h",
+ "exec/common/hash_table/",
+ {
+ # Declarations-only phmap forward header (the sanctioned way
through
+ # the barrier; its name predates the *_fwd.h convention).
+ "exec/common/hash_table/phmap_fwd_decl.h",
+ },
+ "the SharedState classes hold every DataVariants behind unique_ptr/"
+ "shared_ptr with ctors/dtors/close bodies defined in dependency.cpp; "
+ "any path back into the hash-table machinery re-instantiates the "
+ "Agg/Join/Set variant surface (~0.85 CPU s) in each of the ~128 TUs "
+ "that include dependency.h transitively",
+ ),
+ (
+ "exec/pipeline/dependency.h",
+ "exec/operator/join/process_hash_table_probe.h",
+ set(),
+ "dead include: dependency.h references no ProcessHashTableProbe "
+ "symbol; the probe machinery belongs to the hash-join TUs that "
+ "include process_hash_table_probe_impl.h",
+ ),
+ (
+ "exec/pipeline/dependency.h",
+ "util/brpc_closure.h",
+ set(),
+ "dead include: dependency.h references no brpc symbol, yet this edge "
+ "carried runtime/query_context.h, runtime/thread_context.h and "
+ "service/brpc.h (1.36 MB of preprocessed payload) into ~100 TUs "
+ "whose only other route to them was this header",
+ ),
+ (
+ "exec/pipeline/rec_cte_shared_state.h",
+ "exec/common/hash_table/",
+ {
+ "exec/common/hash_table/phmap_fwd_decl.h",
+ },
+ "DistinctDataVariants is forward-declared and only touched in "
+ "rec_cte_shared_state.cpp (emplace_block's std::visit); the distinct "
+ "hash-table family must not ride the rec_cte operator headers into "
+ "the pipeline registry TUs",
+ ),
+ (
+ "exec/pipeline/rec_cte_shared_state.h",
+ "util/brpc_client_cache.h",
+ set(),
+ "send_data_to_targets/build_basic_param bodies live in "
+ "rec_cte_shared_state.cpp; the brpc client stack must not ride a "
+ "SharedState header",
+ ),
]
+# Not expressible as RULES entries (the scanner only follows quoted project
+# includes and <gen_cpp/...>): core/uint24.h and core/value/large_int_value.h
+# must not regain <fmt/compile.h> / <fmt/format.h>. Their to_string/to_buffer
+# bodies live in the matching .cpp files precisely so the FMT_COMPILE formatter
+# templates (53.5 CPU s over ~1150 TUs for the uint24 date format alone) are
+# instantiated once instead of in every includer.
+#
+# Likewise <concurrentqueue.h> must not return to exec/pipeline/dependency.h:
+# it was a dead 152 KB third-party include there; the moodycamel users
+# (local_exchanger.h, scanner_context.h, async_result_writer.h) include it
+# themselves.
+
# Forward-declaration headers are the sanctioned way through a barrier: they
carry
# declarations only, so they cost nothing to include.
FWD_SUFFIX = "_fwd.h"
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
