This is an automated email from the ASF dual-hosted git repository.
HappenLee 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 5b25bbc85b1 [improvement](be) Fuse runtime filter mask merging and
counting with SSE (#69685)
5b25bbc85b1 is described below
commit 5b25bbc85b118ab55a692a9e1d391b35110ad6e8
Author: HappenLee <[email protected]>
AuthorDate: Thu Oct 29 14:27:30 2026 +0800
[improvement](be) Fuse runtime filter mask merging and counting with SSE
(#69685)
### What problem does this PR solve?
Issue Number: N/A
Related PR: N/A
Problem Summary:
`RuntimeFilterExpr::execute_filter` scans the result mask three times
for every non-constant filter: count live input rows, AND the new filter
into the mask, then count live output rows.
Fuse these operations in a shared SSE helper used by both ordinary and
nullable filters. It processes four 16-byte vectors per iteration,
accumulates zero counts in byte lanes (at most four per lane), and
reduces them with `_mm_sad_epu8`. The existing `sse2neon` dependency
maps the same implementation to NEON on AArch64. A scalar tail/fallback
handles remaining rows.
The helper preserves the existing bitwise filtering and nullable
semantics, counts nonzero bytes without assuming the inputs are
normalized to 0/1, and retains the runtime-filter Profile/selectivity
updates.
#### Local microbenchmark
Intel Xeon Platinum 8457C, Clang 21.1.8, `-O3 -DNDEBUG -msse4.2 -mavx2`,
pinned to one CPU. Five repetitions with random interleaving; median CPU
time per batch. The incoming result mask is all ones and the new filter
has 50% passing rows.
| Rows per batch | Before | SSE fusion | CPU time reduction |
|---:|---:|---:|---:|
| 4,096 | 239.57 ns | 188.42 ns | 21.35% |
| 16,384 | 1,014.96 ns | 759.39 ns | 25.18% |
Each iteration restores fresh masks outside the timed region so repeated
AND operations do not gradually turn the inputs into zeros. Independent
blocks use a cache-resident working set; the benchmark measures only
mask merging and counting, not RF evaluation or SQL latency. AArch64
cross-compilation confirms NEON code generation; ARM performance has not
been measured.
### Release note
Reduce CPU overhead when merging runtime-filter masks and counting live
rows.
### Check List (For Author)
- Test
- [ ] Regression test
- [ ] Unit Test
- [x] Manual test
- Added BE unit tests for ordinary/nullable filters, empty batches,
unaligned buffers, SIMD boundaries/tails, and arbitrary byte values; the
test file compiles independently.
- Independent ASAN/UBSan oracle checks passed 11,776 cases for the SSE
path and 11,776 cases for the forced scalar fallback, comparing both
counts and the entire output buffer.
- Ran the SSE microbenchmark matrix for 64/256/1,024/4,093/4,096/16,384
rows and 0/50/100% filter pass rates.
- Cross-compiled the isolated helper through the actual `sse2neon.h` for
AArch64 and inspected NEON code generation. This is not a full ARM BE
build.
- Formatting, header/build hygiene, `git diff --check`, and clang-tidy
on the added helper/test lines passed.
- Attempted `./run-be-ut.sh -j 48 --run
--filter='BitsTest.FilterAndCount*'`; existing local
`ColumnVector`/`PODArray` constructor compilation errors blocked the
full build, so the standard BE unit tests did not execute. Full-TU
clang-tidy is also incomplete because of local compilation/environment
errors.
- Behavior changed:
- [x] No.
- [ ] Yes.
- Does this need documentation?
- [x] No.
- [ ] Yes.
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label
---
be/src/exprs/runtime_filter_expr.cpp | 50 ++++++--------------
be/src/util/simd/bits.h | 59 +++++++++++++++++++++++
be/test/util/simd/bits_test.cpp | 90 ++++++++++++++++++++++++++++++++++++
3 files changed, 164 insertions(+), 35 deletions(-)
diff --git a/be/src/exprs/runtime_filter_expr.cpp
b/be/src/exprs/runtime_filter_expr.cpp
index e491b4329f3..1fa0b1c3708 100644
--- a/be/src/exprs/runtime_filter_expr.cpp
+++ b/be/src/exprs/runtime_filter_expr.cpp
@@ -34,6 +34,7 @@
#include "exprs/vslot_ref.h"
#include "runtime/runtime_profile.h"
#include "storage/index/zone_map/zonemap_eval_context.h"
+#include "util/simd/bits.h"
namespace doris {
@@ -168,52 +169,31 @@ Status RuntimeFilterExpr::execute_filter(VExprContext*
context, const Block* blo
rf_selectivity.update_judge_selectivity(_filter_id, 0, rows,
_ignore_thredhold);
return Status::OK();
}
- } else if (const auto* nullable_column =
check_and_get_column<ColumnNullable>(*filter_column)) {
+ }
+
+ simd::FilterCounts counts;
+ if (const auto* nullable_column =
check_and_get_column<ColumnNullable>(*filter_column)) {
// nullable(bool)
const ColumnPtr& nested_column =
nullable_column->get_nested_column_ptr();
const IColumn::Filter& filter = assert_cast<const
ColumnUInt8&>(*nested_column).get_data();
const auto* __restrict filter_data = filter.data();
const auto* __restrict null_map_data =
nullable_column->get_null_map_data().data();
- const size_t input_rows = rows -
simd::count_zero_num((int8_t*)result_filter_data, rows);
-
- for (size_t i = 0; i < rows; ++i) {
- result_filter_data[i] &= (!null_map_data[i]) & filter_data[i];
- }
-
- const size_t output_rows = rows -
simd::count_zero_num((int8_t*)result_filter_data, rows);
-
- COUNTER_UPDATE(_rf_filter_rows, input_rows - output_rows);
- COUNTER_UPDATE(_rf_input_rows, input_rows);
- rf_selectivity.update_judge_selectivity(_filter_id, input_rows -
output_rows, input_rows,
- _ignore_thredhold);
-
- if (output_rows == 0) {
- *can_filter_all = true;
- return Status::OK();
- }
+ counts = simd::filter_and_count<true>(result_filter_data, filter_data,
rows, null_map_data);
} else {
// bool
const IColumn::Filter& filter = assert_cast<const
ColumnUInt8&>(*filter_column).get_data();
const auto* __restrict filter_data = filter.data();
- const size_t input_rows = rows -
simd::count_zero_num((int8_t*)result_filter_data, rows);
-
- for (size_t i = 0; i < rows; ++i) {
- result_filter_data[i] &= filter_data[i];
- }
-
- const size_t output_rows = rows -
simd::count_zero_num((int8_t*)result_filter_data, rows);
-
- COUNTER_UPDATE(_rf_filter_rows, input_rows - output_rows);
- COUNTER_UPDATE(_rf_input_rows, input_rows);
- rf_selectivity.update_judge_selectivity(_filter_id, input_rows -
output_rows, input_rows,
- _ignore_thredhold);
-
- if (output_rows == 0) {
- *can_filter_all = true;
- return Status::OK();
- }
+ counts = simd::filter_and_count(result_filter_data, filter_data, rows);
+ }
+ const auto [input_rows, output_rows] = counts;
+ COUNTER_UPDATE(_rf_filter_rows, input_rows - output_rows);
+ COUNTER_UPDATE(_rf_input_rows, input_rows);
+ rf_selectivity.update_judge_selectivity(_filter_id, input_rows -
output_rows, input_rows,
+ _ignore_thredhold);
+ if (output_rows == 0) {
+ *can_filter_all = true;
}
return Status::OK();
}
diff --git a/be/src/util/simd/bits.h b/be/src/util/simd/bits.h
index 41b999b0be1..b0bc7c1f654 100644
--- a/be/src/util/simd/bits.h
+++ b/be/src/util/simd/bits.h
@@ -124,6 +124,65 @@ void iterate_through_bits_mask(Func func,
decltype(bytes_mask_to_bits_mask(nullp
#endif
}
+struct FilterCounts {
+ size_t input_rows = 0;
+ size_t output_rows = 0;
+};
+
+// Merge a filter in-place and count nonzero bytes before and after the merge.
+// Nullable filters preserve the scalar expression data[i] &= !null_map[i] &
filter[i].
+template <bool nullable = false>
+inline FilterCounts filter_and_count(uint8_t* __restrict data, const uint8_t*
__restrict filter,
+ size_t size, const uint8_t* __restrict
null_map = nullptr) {
+ FilterCounts counts;
+ size_t i = 0;
+#if defined(__SSE2__) || defined(__aarch64__)
+ const __m128i zero = _mm_setzero_si128();
+ __m128i input_sum = zero;
+ __m128i output_sum = zero;
+ const size_t vector_size = size / 64 * 64;
+ for (; i < vector_size; i += 64) {
+ __m128i input_zeros = zero;
+ __m128i output_zeros = zero;
+ for (size_t offset = 0; offset < 64; offset += 16) {
+ const auto before =
+ _mm_loadu_si128(reinterpret_cast<const __m128i*>(data + i
+ offset));
+ const auto filter_values =
+ _mm_loadu_si128(reinterpret_cast<const __m128i*>(filter +
i + offset));
+ auto after = _mm_and_si128(before, filter_values);
+ if constexpr (nullable) {
+ const auto nulls =
+ _mm_loadu_si128(reinterpret_cast<const
__m128i*>(null_map + i + offset));
+ const auto valid = _mm_and_si128(_mm_cmpeq_epi8(nulls, zero),
_mm_set1_epi8(1));
+ after = _mm_and_si128(after, valid);
+ }
+ // Each byte accumulates at most four zeros. Reduce once per 64
rows
+ // without expanding bytes into size_t lanes or emulating movemask
on ARM.
+ input_zeros = _mm_sub_epi8(input_zeros, _mm_cmpeq_epi8(before,
zero));
+ output_zeros = _mm_sub_epi8(output_zeros, _mm_cmpeq_epi8(after,
zero));
+ _mm_storeu_si128(reinterpret_cast<__m128i*>(data + i + offset),
after);
+ }
+ input_sum = _mm_add_epi64(input_sum, _mm_sad_epu8(input_zeros, zero));
+ output_sum = _mm_add_epi64(output_sum, _mm_sad_epu8(output_zeros,
zero));
+ }
+ counts.input_rows = vector_size -
static_cast<size_t>(_mm_cvtsi128_si64(_mm_add_epi64(
+ input_sum,
_mm_srli_si128(input_sum, 8))));
+ counts.output_rows = vector_size -
static_cast<size_t>(_mm_cvtsi128_si64(_mm_add_epi64(
+ output_sum,
_mm_srli_si128(output_sum, 8))));
+#endif
+ for (; i < size; ++i) {
+ const uint8_t before = data[i];
+ uint8_t after = before & filter[i];
+ if constexpr (nullable) {
+ after &= !null_map[i];
+ }
+ counts.input_rows += before != 0;
+ counts.output_rows += after != 0;
+ data[i] = after;
+ }
+ return counts;
+}
+
template <typename T>
requires requires { std::is_unsigned_v<T>; }
inline T count_zero_num(const int8_t* __restrict data, T size) {
diff --git a/be/test/util/simd/bits_test.cpp b/be/test/util/simd/bits_test.cpp
index abb8e4e340b..4034980baa7 100644
--- a/be/test/util/simd/bits_test.cpp
+++ b/be/test/util/simd/bits_test.cpp
@@ -24,10 +24,100 @@
#include <unistd.h>
#include <algorithm>
+#include <random>
#include "gtest/gtest_pred_impl.h"
namespace doris::simd {
+namespace {
+void fill_filter_inputs(std::vector<uint8_t>& source, std::vector<uint8_t>&
filter,
+ std::vector<uint8_t>& null_map, size_t size, size_t
offset, int pattern,
+ std::mt19937_64& random) {
+ for (size_t i = offset; i < offset + size; ++i) {
+ switch (pattern) {
+ case 0:
+ source[i] = 0;
+ filter[i] = 1;
+ break;
+ case 1:
+ source[i] = 1;
+ filter[i] = 0;
+ break;
+ case 2:
+ source[i] = filter[i] = 1;
+ break;
+ case 3:
+ source[i] = filter[i] = 0xff;
+ break;
+ case 4:
+ source[i] = 0x80;
+ filter[i] = 1;
+ break;
+ case 5:
+ source[i] = i % 2;
+ filter[i] = (i + 1) % 2;
+ break;
+ case 6:
+ source[i] = random() % 2;
+ filter[i] = random() % 2;
+ break;
+ default:
+ source[i] = static_cast<uint8_t>(random());
+ filter[i] = static_cast<uint8_t>(random());
+ break;
+ }
+ null_map[i] = pattern < 3 ? 0 : static_cast<uint8_t>(random() % 3);
+ }
+}
+
+template <bool nullable>
+void check_filter_counts(const std::vector<uint8_t>& source, const
std::vector<uint8_t>& filter,
+ const std::vector<uint8_t>& null_map, size_t size,
size_t offset) {
+ auto expected = source;
+ FilterCounts expected_counts;
+ for (size_t i = offset; i < offset + size; ++i) {
+ expected_counts.input_rows += expected[i] != 0;
+ expected[i] &= filter[i];
+ if constexpr (nullable) {
+ expected[i] &= !null_map[i];
+ }
+ expected_counts.output_rows += expected[i] != 0;
+ }
+ auto actual = source;
+ const auto counts = filter_and_count<nullable>(actual.data() + offset,
filter.data() + offset,
+ size, null_map.data() +
offset);
+ EXPECT_EQ(counts.input_rows, expected_counts.input_rows);
+ EXPECT_EQ(counts.output_rows, expected_counts.output_rows);
+ EXPECT_EQ(actual, expected);
+}
+} // namespace
+
+TEST(BitsTest, FilterAndCount) {
+ std::mt19937_64 random(20261027);
+ for (const size_t size : {0, 1, 7, 15, 16, 17, 31, 32, 33,
63, 64,
+ 65, 127, 128, 129, 255, 256, 257, 1023, 1024,
4093, 4096}) {
+ for (size_t offset = 0; offset < 16; ++offset) {
+ for (int pattern = 0; pattern < 8; ++pattern) {
+ std::vector<uint8_t> source(size + 32, 0xa5);
+ std::vector<uint8_t> filter(size + 32);
+ std::vector<uint8_t> null_map(size + 32);
+ fill_filter_inputs(source, filter, null_map, size, offset,
pattern, random);
+ check_filter_counts<false>(source, filter, null_map, size,
offset);
+ check_filter_counts<true>(source, filter, null_map, size,
offset);
+ }
+ }
+ }
+}
+
+TEST(BitsTest, FilterAndCountEmpty) {
+ const auto counts = filter_and_count(nullptr, nullptr, 0);
+ EXPECT_EQ(counts.input_rows, 0U);
+ EXPECT_EQ(counts.output_rows, 0U);
+ const auto nullable_counts = filter_and_count<true>(nullptr, nullptr, 0,
nullptr);
+ EXPECT_EQ(nullable_counts.input_rows, 0U);
+ EXPECT_EQ(nullable_counts.output_rows, 0U);
+}
+
TEST(BitsTest, ContainsBytesGuardPages) {
const size_t page_size = sysconf(_SC_PAGESIZE);
void* mapping = mmap(nullptr, page_size * 3, PROT_NONE, MAP_PRIVATE |
MAP_ANONYMOUS, -1, 0);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
