This is an automated email from the ASF dual-hosted git repository.
Gabriel39 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new a765b36ebc2 [improvement](parquet) Vectorize File Scanner V2 decode
hot paths (#65972)
a765b36ebc2 is described below
commit a765b36ebc2cb9ea882ae1bdcfb37fedb3a9599b
Author: Gabriel <[email protected]>
AuthorDate: Fri Jul 24 15:59:27 2026 +0800
[improvement](parquet) Vectorize File Scanner V2 decode hot paths (#65972)
### What problem does this PR solve?
File Scanner V2 still executes several fixed-width Parquet decode and
filter stages with scalar row loops. On hot scans, byte-stream-split
transpose, DELTA_BINARY_PACKED reconstruction, numeric dictionary
materialization, nullable expansion, and raw predicate comparison can
become CPU bottlenecks after I/O is cached.
### What is changed?
- Add runtime-dispatched AVX2 kernels with scalar fallbacks for:
- BYTE_STREAM_SPLIT 4-byte and 8-byte transpose
- DELTA_BINARY_PACKED INT32/INT64 prefix reconstruction
- 4-byte and 8-byte numeric dictionary ID gather
- in-place nullable compact-value expansion
- raw INT32/INT64/FLOAT/DOUBLE predicate comparison
- Wire the kernels only into File Scanner V2:
- native byte-stream-split and delta decoders
- cache-resident direct dictionary materialization
- `ColumnChunkReader` nullable POD expansion
- Parquet raw fixed-width predicate evaluation
- Preserve the existing generic paths for unsupported widths, small
batches, non-AVX2 CPUs, and non-cache-resident dictionary strategies.
- Preserve Parquet wrapping arithmetic and Doris floating-point ordering
(`NaN == NaN`, and NaN sorts above finite values).
- Extend the Parquet benchmark matrix:
- decoder selectivity boundaries: 0%, 1%, 10%, 50%, 90%, 100%
- 80 isolated `ParquetKernel` scenarios
- applicable INT32/INT64/FLOAT/DOUBLE physical types
- nullable rates and clustered/alternating null placement
- 32, 4,096, and 262,144 entry dictionary working sets
File Scanner V1 is unchanged.
### Check list
- [x] `ParquetSimdKernelsTest`: 6/6 passed
- [x] `ParquetBenchmarkScenariosTest`: 9/9 passed
- [x] ASAN CMake objects and the Release `benchmark_test` target
compiled and linked
- [x] Real Release `benchmark_main.cpp` compiled with Doris project
flags and `-Werror`
- [x] Release benchmark runner registered 228 decoder, 80 kernel, and
152 reader scenarios
- [x] All 308 decoder and kernel smoke cases passed with zero benchmark
errors
- [x] `git diff --check`
The Release benchmark build and smoke run are execution validation only.
Stable before/after performance comparison should use the checked-in
matrix on a controlled host.
---
be/benchmark/benchmark_main.cpp | 1 +
be/benchmark/parquet/AGENTS.md | 54 +-
be/benchmark/parquet/README.md | 19 +
be/benchmark/parquet/benchmark_parquet_decoder.hpp | 187 +++++-
be/benchmark/parquet/benchmark_parquet_kernels.hpp | 243 ++++++++
be/benchmark/parquet/parquet_benchmark_scenarios.h | 71 +++
.../core/data_type_serde/parquet_decode_source.cpp | 78 +++
.../core/data_type_serde/parquet_decode_source.h | 8 +-
be/src/exprs/vectorized_fn_call.cpp | 33 +-
.../reader/native/byte_stream_split_decoder.cpp | 23 +-
.../parquet/reader/native/column_chunk_reader.cpp | 8 +
.../parquet/reader/native/delta_bit_pack_decoder.h | 11 +-
be/src/util/simd/parquet_kernels.cpp | 651 +++++++++++++++++++++
be/src/util/simd/parquet_kernels.h | 48 ++
be/test/format_v2/parquet/native_decoder_test.cpp | 8 +-
.../parquet/parquet_benchmark_scenarios_test.cpp | 64 +-
.../parquet/parquet_simd_kernels_test.cpp | 222 +++++++
17 files changed, 1654 insertions(+), 75 deletions(-)
diff --git a/be/benchmark/benchmark_main.cpp b/be/benchmark/benchmark_main.cpp
index 582db25ff6d..7c64fa2729c 100644
--- a/be/benchmark/benchmark_main.cpp
+++ b/be/benchmark/benchmark_main.cpp
@@ -45,6 +45,7 @@
#include "core/data_type/data_type.h"
#include "core/data_type/data_type_string.h"
#include "parquet/benchmark_parquet_decoder.hpp"
+#include "parquet/benchmark_parquet_kernels.hpp"
#include "parquet/benchmark_parquet_reader.hpp"
#include "runtime/exec_env.h"
#include "runtime/memory/mem_tracker_limiter.h"
diff --git a/be/benchmark/parquet/AGENTS.md b/be/benchmark/parquet/AGENTS.md
index a558f69937c..ba52658a966 100644
--- a/be/benchmark/parquet/AGENTS.md
+++ b/be/benchmark/parquet/AGENTS.md
@@ -6,9 +6,10 @@ benchmark system described in the design document.
## What exists today
-The benchmark binary registers two groups:
+The benchmark binary registers three groups:
- `ParquetDecoder`: native page decoder benchmarks using in-memory encoded
pages.
+- `ParquetKernel`: isolated SIMD-sensitive decode and predicate kernels.
- `ParquetReader`: local-file benchmarks that call the format V2 Parquet
reader directly.
The relevant files are:
@@ -36,10 +37,13 @@ List all Parquet cases and verify the expected registration
counts:
```shell
be/output/lib/benchmark_test --benchmark_list_tests \
- | grep -E '^Parquet(Decoder|Reader)/'
+ | grep -E '^Parquet(Decoder|Kernel|Reader)/'
be/output/lib/benchmark_test --benchmark_list_tests \
- | grep -c '^ParquetDecoder/' # currently 152
+ | grep -c '^ParquetDecoder/' # currently 228
+
+be/output/lib/benchmark_test --benchmark_list_tests \
+ | grep -c '^ParquetKernel/' # currently 80
be/output/lib/benchmark_test --benchmark_list_tests \
| grep -c '^ParquetReader/' # currently 152
@@ -61,6 +65,12 @@ be/output/lib/benchmark_test \
--benchmark_out=parquet-decoder-smoke.json \
--benchmark_out_format=json
+be/output/lib/benchmark_test \
+ --benchmark_filter='^ParquetKernel/' \
+ --benchmark_min_time=0.001s \
+ --benchmark_out=parquet-kernel-smoke.json \
+ --benchmark_out_format=json
+
be/output/lib/benchmark_test \
--benchmark_filter='^ParquetReader/' \
--benchmark_min_time=0.001s \
@@ -98,8 +108,8 @@ cache to manufacture a cold run.
## Current scenario matrix
-`ParquetDecoder` contains 19 encoding/type pairs. Each pair is run at 1%, 10%,
50%, and 100%
-selection with clustered and alternating selection ranges, for 152 registered
cases.
+`ParquetDecoder` contains 19 encoding/type pairs. Each pair is run at 0%, 1%,
10%, 50%, 90%, and
+100% selection with clustered and alternating selection ranges, for 228
registered cases.
| Encoding | Physical types |
|---|---|
@@ -110,6 +120,11 @@ selection with clustered and alternating selection ranges,
for 152 registered ca
| DELTA_LENGTH_BYTE_ARRAY | BYTE_ARRAY |
| DELTA_BYTE_ARRAY | BYTE_ARRAY |
+`ParquetKernel` contains 80 cases across five SIMD-sensitive stages:
BYTE_STREAM_SPLIT,
+DELTA_PREFIX_SUM, DICTIONARY_GATHER, NULLABLE_EXPAND, and RAW_PREDICATE. It
covers the applicable
+four- and eight-byte types, three dictionary working-set sizes, 0% through 90%
null rates with both
+placement patterns, and 0% through 100% raw-predicate selectivities.
+
`ParquetReader` deliberately uses a single-variable matrix rather than a
Cartesian product. After
deduplication it contains 152 cases covering:
@@ -147,10 +162,12 @@ generator, random seed, or manifest involved.
and intentionally produces many one-row physical ranges.
Page generation, selection construction, decoder creation, dictionary setup,
and `set_data` are
-outside the timed decode call. The sinks consume decoder callbacks and prevent
compiler removal,
-but they do not build a Doris `Column`. Consequently these cases isolate
decoder traversal and
-selection cost; they do not measure definition-level decoding, nullable
reconstruction, type
-conversion, or full column materialization.
+outside the timed decode call. Before timing, every decoder case verifies the
consumed value count
+and a checksum of all selected values against the deterministic source
generator. The timed sinks
+then consume decoder callbacks and prevent compiler removal, but they do not
build a Doris
+`Column`. Consequently these cases isolate decoder traversal and selection
cost; they do not
+measure definition-level decoding, nullable reconstruction, type conversion,
or full column
+materialization.
## How reader Parquet files are generated
@@ -238,13 +255,14 @@ The current matrix is not comprehensive. Preserve this
distinction in PR descrip
1. Add the design's `matrix.yaml`, deterministic corpus generator,
`manifest.json`, checksum, and
standalone corpus verifier. The current runtime-generated files cannot be
shared unchanged with
V1, StarRocks, or DuckDB.
-2. Add correctness oracles. Benchmarks must validate consumed counts and
representative checksums
- outside the timed region before their performance samples are trusted.
+2. Add full reader correctness oracles. Decoder cases validate consumed counts
and selected-value
+ checksums outside the timed region, and kernel cases compare representative
output. Reader cases
+ still need value checksums in addition to their output-row counters.
3. Add reader-level INT64, FLOAT, DOUBLE, BYTE_ARRAY/string,
FIXED_LEN_BYTE_ARRAY, DATE,
TIMESTAMP, and DECIMAL cases. Today only nullable INT32 reaches the
complete reader path.
-4. Extend decoder coverage with 0% and 90% selection, definition levels/null
reconstruction,
- dictionary conversion, and real Doris `Column` materialization. The current
decoder sink does
- not cover those costs or report decoded bytes per second.
+4. Extend decoder coverage with definition levels/null reconstruction,
dictionary conversion, and
+ real Doris `Column` materialization. The current decoder sink does not
cover those costs or
+ report decoded bytes per second.
5. Add the representative nullable sparse corpus requested by the design: 32
INT64 columns, 128
row groups, and enough rows to exercise many pages. The current
16K-row/four-row-group fixture
is a smoke-sized workload.
@@ -281,10 +299,10 @@ be simulated by silently changing the local reader
benchmark.
## Current validation record
-At commit `16e05dd5c71`, a Release build completed and the matrix unit test
passed 4/4. A 1 ms smoke
-run executed 152 decoder and 137 reader cases with zero benchmark errors. This
is an execution
-record only. It is not a reviewed performance baseline because repetitions,
host isolation,
-warmups, cache control, `perf` data, variance, and before/after comparison
were not collected.
+The current expected registration counts are 228 decoder, 80 kernel, and 152
reader cases. A smoke
+run is an execution record only, not a reviewed performance baseline, because
repetitions, host
+isolation, warmups, cache control, `perf` data, variance, and before/after
comparison are not
+collected.
## Rules for extending the suite
diff --git a/be/benchmark/parquet/README.md b/be/benchmark/parquet/README.md
index f024f7d6a24..e6eab6362ce 100644
--- a/be/benchmark/parquet/README.md
+++ b/be/benchmark/parquet/README.md
@@ -28,12 +28,31 @@ timed region. It covers PLAIN, dictionary,
byte-stream-split, and DELTA encoding
supported fixed-width and binary physical types. Sparse selections are
provided as both one
clustered range and many alternating ranges.
+The decoder selection axis includes 0%, 1%, 10%, 50%, 90%, and 100% so
boundary and
+high-selectivity behavior are visible.
+
```shell
be/output/lib/benchmark_test \
--benchmark_filter='^ParquetDecoder/plain/int64/sel_10/alternating$' \
--benchmark_min_time=0.1s
```
+## SIMD kernel cases
+
+`ParquetKernel` isolates the five SIMD-sensitive stages from reader setup and
virtual consumer
+overhead: byte-stream-split transpose, delta prefix sum, numeric dictionary
gather, nullable
+expansion, and raw predicate evaluation. It covers the applicable 4-byte and
8-byte integer and
+floating-point physical types, raw-predicate selectivities from 0% through
100%, and nullable
+rates from 0% through 90% with clustered and alternating placement. Dictionary
gather uses 32-,
+4,096-, and 262,144-entry working sets to separate cache-resident and
cache-miss-dominated
+behavior.
+
+```shell
+be/output/lib/benchmark_test \
+ --benchmark_filter='^ParquetKernel/(dictionary_gather|nullable_expand)/' \
+ --benchmark_min_time=0.1s
+```
+
## Local reader cases
`ParquetReader` measures local open-to-first-block, full scan, predicate scan,
complex residual
diff --git a/be/benchmark/parquet/benchmark_parquet_decoder.hpp
b/be/benchmark/parquet/benchmark_parquet_decoder.hpp
index 25fb05d1449..c1c2e3cb332 100644
--- a/be/benchmark/parquet/benchmark_parquet_decoder.hpp
+++ b/be/benchmark/parquet/benchmark_parquet_decoder.hpp
@@ -101,31 +101,87 @@ inline std::shared_ptr<::parquet::ColumnDescriptor>
descriptor(::parquet::Type::
return std::make_shared<::parquet::ColumnDescriptor>(node, 0, 0);
}
+template <typename T>
+T fixed_value(size_t row) {
+ if constexpr (std::is_floating_point_v<T>) {
+ return static_cast<T>((row % 1009) * 0.25 - 100.0);
+ }
+ return static_cast<T>((row * 17) % 1000003);
+}
+
template <typename T>
std::vector<T> fixed_values(size_t rows) {
std::vector<T> values(rows);
for (size_t row = 0; row < rows; ++row) {
- if constexpr (std::is_floating_point_v<T>) {
- values[row] = static_cast<T>((row % 1009) * 0.25 - 100.0);
- } else {
- values[row] = static_cast<T>((row * 17) % 1000003);
- }
+ values[row] = fixed_value<T>(row);
}
return values;
}
+inline std::string binary_value(size_t row, size_t width = FIXED_BINARY_WIDTH)
{
+ std::string value(width, 'a');
+ const uint64_t id = row % 1009;
+ memcpy(value.data(), &id, std::min(width, sizeof(id)));
+ return value;
+}
+
inline std::vector<std::string> binary_values(size_t rows, size_t width =
FIXED_BINARY_WIDTH) {
std::vector<std::string> values;
values.reserve(rows);
for (size_t row = 0; row < rows; ++row) {
- std::string value(width, 'a');
- const uint64_t id = row % 1009;
- memcpy(value.data(), &id, std::min(width, sizeof(id)));
- values.push_back(std::move(value));
+ values.push_back(binary_value(row, width));
}
return values;
}
+struct DecoderDigest {
+ size_t consumed = 0;
+ uint64_t checksum = 1469598103934665603ULL;
+};
+
+inline void add_digest_value(DecoderDigest* digest, const uint8_t* bytes,
size_t size) {
+ constexpr uint64_t FNV_PRIME = 1099511628211ULL;
+ digest->checksum ^= size;
+ digest->checksum *= FNV_PRIME;
+ for (size_t byte = 0; byte < size; ++byte) {
+ digest->checksum ^= bytes[byte];
+ digest->checksum *= FNV_PRIME;
+ }
+ ++digest->consumed;
+}
+
+inline void add_generated_value(DecoderDigest* digest, ValueType value_type,
size_t row) {
+ switch (value_type) {
+ case ValueType::INT32: {
+ const auto value = fixed_value<int32_t>(row);
+ add_digest_value(digest, reinterpret_cast<const uint8_t*>(&value),
sizeof(value));
+ return;
+ }
+ case ValueType::INT64: {
+ const auto value = fixed_value<int64_t>(row);
+ add_digest_value(digest, reinterpret_cast<const uint8_t*>(&value),
sizeof(value));
+ return;
+ }
+ case ValueType::FLOAT: {
+ const auto value = fixed_value<float>(row);
+ add_digest_value(digest, reinterpret_cast<const uint8_t*>(&value),
sizeof(value));
+ return;
+ }
+ case ValueType::DOUBLE: {
+ const auto value = fixed_value<double>(row);
+ add_digest_value(digest, reinterpret_cast<const uint8_t*>(&value),
sizeof(value));
+ return;
+ }
+ case ValueType::BYTE_ARRAY:
+ case ValueType::FIXED_LEN_BYTE_ARRAY: {
+ const auto value = binary_value(row);
+ add_digest_value(digest, reinterpret_cast<const
uint8_t*>(value.data()), value.size());
+ return;
+ }
+ }
+ throw std::logic_error("unknown Parquet benchmark value type");
+}
+
inline std::vector<uint8_t> encode_plain_binary(const
std::vector<std::string>& values) {
size_t bytes = 0;
for (const auto& value : values) {
@@ -376,6 +432,67 @@ private:
const size_t _value_width;
};
+class FixedVerificationSink final : public ParquetFixedValueConsumer {
+public:
+ Status consume(const uint8_t* values, size_t num_values, size_t
value_width) override {
+ for (size_t row = 0; row < num_values; ++row) {
+ add_digest_value(&digest, values + row * value_width, value_width);
+ }
+ return Status::OK();
+ }
+
+ DecoderDigest digest;
+};
+
+class BinaryVerificationSink final : public ParquetBinaryValueConsumer {
+public:
+ Status consume(const StringRef* values, size_t num_values) override {
+ for (size_t row = 0; row < num_values; ++row) {
+ add_digest_value(&digest, reinterpret_cast<const
uint8_t*>(values[row].data),
+ values[row].size);
+ }
+ return Status::OK();
+ }
+
+ Status consume_plain_byte_array(
+ const char* encoded_data, const uint32_t* payload_offsets,
+ const uint32_t* value_offsets, size_t num_values,
+ const std::vector<ParquetSelectionRange>& value_spans) override {
+ for (size_t row = 0; row < num_values; ++row) {
+ add_digest_value(&digest,
+ reinterpret_cast<const uint8_t*>(encoded_data +
payload_offsets[row]),
+ value_offsets[row + 1] - value_offsets[row]);
+ }
+ return Status::OK();
+ }
+
+ DecoderDigest digest;
+};
+
+class DictionaryVerificationSink final : public ParquetDictionaryValueConsumer
{
+public:
+ explicit DictionaryVerificationSink(ValueType value_type) :
_value_type(value_type) {}
+
+ Status consume_indices(const uint32_t* indices, size_t num_values)
override {
+ for (size_t row = 0; row < num_values; ++row) {
+ add_generated_value(&digest, _value_type, indices[row]);
+ }
+ return Status::OK();
+ }
+
+ Status consume_repeated(uint32_t index, size_t num_values) override {
+ for (size_t row = 0; row < num_values; ++row) {
+ add_generated_value(&digest, _value_type, index);
+ }
+ return Status::OK();
+ }
+
+ DecoderDigest digest;
+
+private:
+ const ValueType _value_type;
+};
+
inline ParquetSelection native_selection(const SelectionPlan& plan) {
ParquetSelection selection {
.total_values = plan.total_rows, .selected_values =
plan.selected_rows, .ranges = {}};
@@ -386,6 +503,50 @@ inline ParquetSelection native_selection(const
SelectionPlan& plan) {
return selection;
}
+inline DecoderDigest expected_decoder_digest(const DecoderScenario& scenario,
+ const SelectionPlan& plan) {
+ DecoderDigest digest;
+ visit_selected_rows(plan, [&](size_t row) {
+ const size_t value_row =
+ scenario.encoding == Encoding::DICTIONARY ? row %
DICTIONARY_ENTRIES : row;
+ add_generated_value(&digest, scenario.value_type, value_row);
+ });
+ return digest;
+}
+
+inline Status verify_decoder_output(format::parquet::native::Decoder* decoder,
Slice* encoded,
+ const DecoderScenario& scenario, bool
binary,
+ const ParquetSelection& selection,
+ const SelectionPlan& plan) {
+ RETURN_IF_ERROR(decoder->set_data(encoded));
+ DecoderDigest actual;
+ if (scenario.encoding == Encoding::DICTIONARY) {
+ DictionaryVerificationSink sink(scenario.value_type);
+ RETURN_IF_ERROR(decoder->decode_selected_dictionary_values(selection,
sink));
+ actual = sink.digest;
+ } else if (binary) {
+ BinaryVerificationSink sink;
+ RETURN_IF_ERROR(decoder->decode_selected_binary_values(selection,
sink));
+ actual = sink.digest;
+ } else {
+ FixedVerificationSink sink;
+ RETURN_IF_ERROR(decoder->decode_selected_fixed_values(selection,
sink));
+ actual = sink.digest;
+ }
+
+ const auto expected = expected_decoder_digest(scenario, plan);
+ // Validate sparse/boundary selections before timing; counters alone can
hide wrong selected
+ // rows while still producing a plausible benchmark result.
+ if (actual.consumed != plan.selected_rows || actual.consumed !=
expected.consumed ||
+ actual.checksum != expected.checksum) {
+ return Status::InternalError(
+ "Parquet decoder benchmark oracle mismatch: consumed {}
expected {}, checksum {} "
+ "expected {}",
+ actual.consumed, plan.selected_rows, actual.checksum,
expected.checksum);
+ }
+ return Status::OK();
+}
+
inline void run_decoder(benchmark::State& state, DecoderScenario scenario, int
selectivity,
Pattern pattern) {
auto page = encoded_page(scenario);
@@ -415,6 +576,12 @@ inline void run_decoder(benchmark::State& state,
DecoderScenario scenario, int s
}
Slice encoded(page.data.data(), page.data.size());
+ status = verify_decoder_output(decoder.get(), &encoded, scenario,
page.binary, selection, plan);
+ if (!status.ok()) {
+ state.SkipWithError(status.to_string().c_str());
+ return;
+ }
+
FixedSink fixed_sink;
BinarySink binary_sink;
DictionarySink dictionary_sink(dictionary_for_sink.data(),
page.value_width);
@@ -459,7 +626,7 @@ inline void run_decoder(benchmark::State& state,
DecoderScenario scenario, int s
inline bool register_decoder_benchmarks() {
for (const auto& scenario : decoder_scenarios()) {
- for (const int selectivity : {1, 10, 50, 100}) {
+ for (const int selectivity : {0, 1, 10, 50, 90, 100}) {
for (const auto pattern : {Pattern::CLUSTERED,
Pattern::ALTERNATING}) {
const std::string name = "ParquetDecoder/" +
to_string(scenario.encoding) + "/" +
to_string(scenario.value_type) +
"/sel_" +
diff --git a/be/benchmark/parquet/benchmark_parquet_kernels.hpp
b/be/benchmark/parquet/benchmark_parquet_kernels.hpp
new file mode 100644
index 00000000000..e064e69f64f
--- /dev/null
+++ b/be/benchmark/parquet/benchmark_parquet_kernels.hpp
@@ -0,0 +1,243 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#pragma once
+
+#include <benchmark/benchmark.h>
+
+#include <algorithm>
+#include <cstddef>
+#include <cstdint>
+#include <string>
+#include <type_traits>
+#include <vector>
+
+#include "parquet_benchmark_scenarios.h"
+#include "util/byte_stream_split.h"
+#include "util/simd/parquet_kernels.h"
+
+namespace doris::parquet_benchmark {
+namespace detail {
+
+constexpr size_t KERNEL_ROWS = 1UL << 16;
+
+inline void decode_byte_stream_split(const uint8_t* src, size_t width, size_t
offset,
+ size_t num_values, size_t stride,
uint8_t* dest) {
+ if (!simd::try_byte_stream_split_decode(src, width, offset, num_values,
stride, dest)) {
+ doris::byte_stream_split_decode(src, static_cast<int>(width), offset,
num_values, stride,
+ dest);
+ }
+}
+
+template <typename T>
+void run_kernel(benchmark::State& state, const KernelScenario& scenario) {
+ constexpr size_t width = sizeof(T);
+ std::vector<T> input(KERNEL_ROWS);
+ for (size_t row = 0; row < input.size(); ++row) {
+ input[row] = scenario.kernel == Kernel::RAW_PREDICATE ?
static_cast<T>(row % 100)
+ :
static_cast<T>((row * 17) % 1009);
+ }
+ std::vector<T> output(KERNEL_ROWS);
+
+ std::vector<uint8_t> encoded(KERNEL_ROWS * width);
+ for (size_t row = 0; row < KERNEL_ROWS; ++row) {
+ for (size_t byte = 0; byte < width; ++byte) {
+ encoded[byte * KERNEL_ROWS + row] =
+ reinterpret_cast<const uint8_t*>(input.data())[row * width
+ byte];
+ }
+ }
+
+ std::vector<T> dictionary(scenario.dictionary_entries);
+ for (size_t row = 0; row < dictionary.size(); ++row) {
+ dictionary[row] = static_cast<T>(row * 31);
+ }
+ std::vector<uint32_t> ids(KERNEL_ROWS);
+ for (size_t row = 0; row < ids.size(); ++row) {
+ ids[row] = static_cast<uint32_t>((row * 13) % dictionary.size());
+ }
+
+ std::vector<uint8_t> nulls(KERNEL_ROWS, 0);
+ const size_t null_count = KERNEL_ROWS *
static_cast<size_t>(scenario.null_percent) / 100;
+ if (scenario.pattern == Pattern::CLUSTERED) {
+ std::fill_n(nulls.begin(), null_count, uint8_t {1});
+ } else if (null_count != 0) {
+ for (size_t value = 0; value < null_count; ++value) {
+ nulls[value * KERNEL_ROWS / null_count] = 1;
+ }
+ }
+ std::vector<T> compact;
+ compact.reserve(KERNEL_ROWS - null_count);
+ for (size_t row = 0; row < KERNEL_ROWS; ++row) {
+ if (nulls[row] == 0) {
+ compact.push_back(input[row]);
+ }
+ }
+
+ const T literal = static_cast<T>(scenario.selectivity_percent);
+ std::vector<uint8_t> matches(KERNEL_ROWS, 1);
+
+ switch (scenario.kernel) {
+ case Kernel::BYTE_STREAM_SPLIT:
+ decode_byte_stream_split(encoded.data(), width, 0, KERNEL_ROWS,
KERNEL_ROWS,
+ reinterpret_cast<uint8_t*>(output.data()));
+ if (output != input) {
+ state.SkipWithError("byte-stream-split kernel produced incorrect
values");
+ return;
+ }
+ break;
+ case Kernel::DELTA_PREFIX_SUM: {
+ if constexpr (std::is_integral_v<T>) {
+ std::copy(input.begin(), input.end(), output.begin());
+ auto expected = output;
+ T expected_last = 7;
+ using Unsigned = std::make_unsigned_t<T>;
+ for (auto& value : expected) {
+ value = static_cast<T>(static_cast<Unsigned>(value) +
+
static_cast<Unsigned>(static_cast<T>(-3)) +
+ static_cast<Unsigned>(expected_last));
+ expected_last = value;
+ }
+ T last = 7;
+ simd::delta_decode(output.data(), output.size(),
static_cast<T>(-3), &last);
+ if (output != expected || last != expected_last) {
+ state.SkipWithError("delta prefix-sum kernel produced
incorrect values");
+ return;
+ }
+ } else {
+ state.SkipWithError("delta prefix-sum requires an integer physical
type");
+ return;
+ }
+ break;
+ }
+ case Kernel::DICTIONARY_GATHER:
+ simd::dictionary_gather(reinterpret_cast<const
uint8_t*>(dictionary.data()), ids.data(),
+ ids.size(), width,
reinterpret_cast<uint8_t*>(output.data()));
+ for (size_t row = 0; row < output.size(); ++row) {
+ if (output[row] != dictionary[ids[row]]) {
+ state.SkipWithError("dictionary gather kernel produced
incorrect values");
+ return;
+ }
+ }
+ break;
+ case Kernel::NULLABLE_EXPAND:
+ std::copy(compact.begin(), compact.end(), output.begin());
+
simd::expand_nullable_values(reinterpret_cast<uint8_t*>(output.data()),
compact.size(),
+ nulls.data(), nulls.size(), width);
+ for (size_t row = 0; row < output.size(); ++row) {
+ const T expected = nulls[row] == 0 ? input[row] : T {};
+ if (output[row] != expected) {
+ state.SkipWithError("nullable expansion kernel produced
incorrect values");
+ return;
+ }
+ }
+ break;
+ case Kernel::RAW_PREDICATE:
+ simd::raw_compare(reinterpret_cast<const uint8_t*>(input.data()),
input.size(), literal,
+ simd::RawComparisonOp::LT, matches.data());
+ for (size_t row = 0; row < matches.size(); ++row) {
+ if (matches[row] != static_cast<uint8_t>(input[row] < literal)) {
+ state.SkipWithError("raw predicate kernel produced incorrect
values");
+ return;
+ }
+ }
+ break;
+ }
+
+ for (auto _ : state) {
+ state.PauseTiming();
+ if (scenario.kernel == Kernel::DELTA_PREFIX_SUM) {
+ std::copy(input.begin(), input.end(), output.begin());
+ } else if (scenario.kernel == Kernel::NULLABLE_EXPAND) {
+ std::copy(compact.begin(), compact.end(), output.begin());
+ } else if (scenario.kernel == Kernel::RAW_PREDICATE) {
+ std::fill(matches.begin(), matches.end(), uint8_t {1});
+ }
+ state.ResumeTiming();
+
+ switch (scenario.kernel) {
+ case Kernel::BYTE_STREAM_SPLIT:
+ decode_byte_stream_split(encoded.data(), width, 0, KERNEL_ROWS,
KERNEL_ROWS,
+
reinterpret_cast<uint8_t*>(output.data()));
+ break;
+ case Kernel::DELTA_PREFIX_SUM: {
+ if constexpr (std::is_integral_v<T>) {
+ T last = 7;
+ simd::delta_decode(output.data(), output.size(),
static_cast<T>(-3), &last);
+ benchmark::DoNotOptimize(last);
+ }
+ break;
+ }
+ case Kernel::DICTIONARY_GATHER:
+ simd::dictionary_gather(reinterpret_cast<const
uint8_t*>(dictionary.data()), ids.data(),
+ ids.size(), width,
reinterpret_cast<uint8_t*>(output.data()));
+ break;
+ case Kernel::NULLABLE_EXPAND:
+
simd::expand_nullable_values(reinterpret_cast<uint8_t*>(output.data()),
compact.size(),
+ nulls.data(), nulls.size(), width);
+ break;
+ case Kernel::RAW_PREDICATE:
+ simd::raw_compare(reinterpret_cast<const uint8_t*>(input.data()),
input.size(), literal,
+ simd::RawComparisonOp::LT, matches.data());
+ break;
+ }
+ benchmark::ClobberMemory();
+ }
+
+ state.SetItemsProcessed(static_cast<int64_t>(state.iterations()) *
+ static_cast<int64_t>(KERNEL_ROWS));
+ state.SetBytesProcessed(static_cast<int64_t>(state.iterations()) *
+ static_cast<int64_t>(KERNEL_ROWS * width));
+ state.counters["rows"] = static_cast<double>(KERNEL_ROWS);
+ state.counters["value_width"] = static_cast<double>(width);
+}
+
+inline bool register_kernel_benchmarks() {
+ for (const auto& scenario : kernel_scenarios()) {
+ const std::string name = "ParquetKernel/" + to_string(scenario.kernel)
+ "/" +
+ to_string(scenario.value_type) + "/sel_" +
+ std::to_string(scenario.selectivity_percent)
+ "/null_" +
+ std::to_string(scenario.null_percent) + "/" +
+ to_string(scenario.pattern) + "/dict_" +
+ std::to_string(scenario.dictionary_entries);
+ benchmark::RegisterBenchmark(name.c_str(), [=](benchmark::State&
state) {
+ switch (scenario.value_type) {
+ case ValueType::INT32:
+ run_kernel<int32_t>(state, scenario);
+ break;
+ case ValueType::INT64:
+ run_kernel<int64_t>(state, scenario);
+ break;
+ case ValueType::FLOAT:
+ run_kernel<float>(state, scenario);
+ break;
+ case ValueType::DOUBLE:
+ run_kernel<double>(state, scenario);
+ break;
+ case ValueType::BYTE_ARRAY:
+ case ValueType::FIXED_LEN_BYTE_ARRAY:
+ state.SkipWithError("kernel benchmark requires a fixed-width
primitive type");
+ break;
+ }
+ })->Unit(benchmark::kNanosecond);
+ }
+ return true;
+}
+
+inline const bool KERNEL_BENCHMARKS_REGISTERED = register_kernel_benchmarks();
+
+} // namespace detail
+} // namespace doris::parquet_benchmark
diff --git a/be/benchmark/parquet/parquet_benchmark_scenarios.h
b/be/benchmark/parquet/parquet_benchmark_scenarios.h
index 1db6b3c8fd2..e6ef9f367be 100644
--- a/be/benchmark/parquet/parquet_benchmark_scenarios.h
+++ b/be/benchmark/parquet/parquet_benchmark_scenarios.h
@@ -45,6 +45,13 @@ enum class ReaderOperation {
LIMIT_1,
LIMIT_1000
};
+enum class Kernel {
+ BYTE_STREAM_SPLIT,
+ DELTA_PREFIX_SUM,
+ DICTIONARY_GATHER,
+ NULLABLE_EXPAND,
+ RAW_PREDICATE
+};
struct DecoderScenario {
Encoding encoding;
@@ -62,6 +69,15 @@ struct ReaderScenario {
int predicate_position;
};
+struct KernelScenario {
+ Kernel kernel;
+ ValueType value_type;
+ int selectivity_percent;
+ int null_percent;
+ Pattern pattern;
+ size_t dictionary_entries;
+};
+
struct SelectionRange {
size_t first;
size_t count;
@@ -97,6 +113,36 @@ inline std::vector<DecoderScenario> decoder_scenarios() {
};
}
+inline std::vector<KernelScenario> kernel_scenarios() {
+ std::vector<KernelScenario> scenarios;
+ for (const auto value_type : {ValueType::FLOAT, ValueType::DOUBLE}) {
+ scenarios.push_back(
+ {Kernel::BYTE_STREAM_SPLIT, value_type, 100, 0,
Pattern::CLUSTERED, 256});
+ }
+ for (const auto value_type : {ValueType::INT32, ValueType::INT64}) {
+ scenarios.push_back(
+ {Kernel::DELTA_PREFIX_SUM, value_type, 100, 0,
Pattern::CLUSTERED, 256});
+ }
+ for (const auto value_type :
+ {ValueType::INT32, ValueType::INT64, ValueType::FLOAT,
ValueType::DOUBLE}) {
+ for (const size_t dictionary_entries : {32, 4096, 262144}) {
+ scenarios.push_back({Kernel::DICTIONARY_GATHER, value_type, 100,
0, Pattern::CLUSTERED,
+ dictionary_entries});
+ }
+ for (const int null_percent : {0, 1, 10, 50, 90}) {
+ for (const auto pattern : {Pattern::CLUSTERED,
Pattern::ALTERNATING}) {
+ scenarios.push_back(
+ {Kernel::NULLABLE_EXPAND, value_type, 100,
null_percent, pattern, 256});
+ }
+ }
+ for (const int selectivity : {0, 1, 10, 50, 90, 100}) {
+ scenarios.push_back(
+ {Kernel::RAW_PREDICATE, value_type, selectivity, 0,
Pattern::ALTERNATING, 256});
+ }
+ }
+ return scenarios;
+}
+
inline std::vector<ReaderScenario> reader_scenarios() {
std::vector<ReaderScenario> scenarios;
std::set<std::tuple<ReaderOperation, Encoding, int, Pattern, int,
Projection, int, int>> seen;
@@ -208,6 +254,15 @@ inline SelectionPlan make_selection_plan(size_t
total_rows, int selectivity_perc
return plan;
}
+template <typename Visitor>
+inline void visit_selected_rows(const SelectionPlan& plan, Visitor visitor) {
+ for (const auto& range : plan.ranges) {
+ for (size_t offset = 0; offset < range.count; ++offset) {
+ visitor(range.first + offset);
+ }
+ }
+}
+
inline std::string to_string(Encoding value) {
switch (value) {
case Encoding::PLAIN:
@@ -278,4 +333,20 @@ inline std::string reader_scenario_name(const
ReaderScenario& scenario) {
"/predicate_" + std::to_string(scenario.predicate_position);
}
+inline std::string to_string(Kernel value) {
+ switch (value) {
+ case Kernel::BYTE_STREAM_SPLIT:
+ return "byte_stream_split";
+ case Kernel::DELTA_PREFIX_SUM:
+ return "delta_prefix_sum";
+ case Kernel::DICTIONARY_GATHER:
+ return "dictionary_gather";
+ case Kernel::NULLABLE_EXPAND:
+ return "nullable_expand";
+ case Kernel::RAW_PREDICATE:
+ return "raw_predicate";
+ }
+ return "unknown";
+}
+
} // namespace doris::parquet_benchmark
diff --git a/be/src/core/data_type_serde/parquet_decode_source.cpp
b/be/src/core/data_type_serde/parquet_decode_source.cpp
new file mode 100644
index 00000000000..8a60288811c
--- /dev/null
+++ b/be/src/core/data_type_serde/parquet_decode_source.cpp
@@ -0,0 +1,78 @@
+// 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 "core/data_type_serde/parquet_decode_source.h"
+
+#include "core/column/column_vector.h"
+#include "util/simd/parquet_kernels.h"
+
+namespace doris {
+namespace {
+
+template <PrimitiveType TYPE>
+bool try_gather_vector(IColumn& destination, const IColumn& dictionary, const
uint32_t* indices,
+ size_t num_values) {
+ using ColumnType = ColumnVector<TYPE>;
+ using ValueType = typename ColumnType::value_type;
+ if constexpr (sizeof(ValueType) != 4 && sizeof(ValueType) != 8) {
+ return false;
+ } else {
+ auto* destination_vector = dynamic_cast<ColumnType*>(&destination);
+ const auto* dictionary_vector = dynamic_cast<const
ColumnType*>(&dictionary);
+ if (destination_vector == nullptr || dictionary_vector == nullptr) {
+ return false;
+ }
+ // The direct strategy is chosen only while the typed dictionary is
cache-resident. Keep
+ // the existing generic insertion path for tiny batches where gather
setup cannot amortize.
+ constexpr size_t SIMD_LANES = sizeof(ValueType) == 4 ? 8 : 4;
+ if (num_values < SIMD_LANES) {
+ return false;
+ }
+ auto& destination_data = destination_vector->get_data();
+ const auto& dictionary_data = dictionary_vector->get_data();
+ const size_t old_size = destination_data.size();
+ destination_data.resize(old_size + num_values);
+ simd::dictionary_gather(reinterpret_cast<const
uint8_t*>(dictionary_data.data()), indices,
+ num_values, sizeof(ValueType),
+
reinterpret_cast<uint8_t*>(destination_data.data() + old_size));
+ return true;
+ }
+}
+
+} // namespace
+
+bool try_simd_insert_parquet_dictionary_indices(IColumn& destination, const
IColumn& dictionary,
+ const uint32_t* indices,
size_t num_values) {
+#define TRY_PARQUET_GATHER(TYPE) \
+ if (try_gather_vector<TYPE>(destination, dictionary, indices, num_values))
return true
+ TRY_PARQUET_GATHER(TYPE_INT);
+ TRY_PARQUET_GATHER(TYPE_BIGINT);
+ TRY_PARQUET_GATHER(TYPE_FLOAT);
+ TRY_PARQUET_GATHER(TYPE_DOUBLE);
+ TRY_PARQUET_GATHER(TYPE_DATE);
+ TRY_PARQUET_GATHER(TYPE_DATETIME);
+ TRY_PARQUET_GATHER(TYPE_DATEV2);
+ TRY_PARQUET_GATHER(TYPE_DATETIMEV2);
+ TRY_PARQUET_GATHER(TYPE_IPV4);
+ TRY_PARQUET_GATHER(TYPE_TIMEV2);
+ TRY_PARQUET_GATHER(TYPE_UINT32);
+ TRY_PARQUET_GATHER(TYPE_UINT64);
+#undef TRY_PARQUET_GATHER
+ return false;
+}
+
+} // namespace doris
diff --git a/be/src/core/data_type_serde/parquet_decode_source.h
b/be/src/core/data_type_serde/parquet_decode_source.h
index e9762385689..18624abf45e 100644
--- a/be/src/core/data_type_serde/parquet_decode_source.h
+++ b/be/src/core/data_type_serde/parquet_decode_source.h
@@ -253,6 +253,9 @@ public:
enum class ParquetDictionaryMaterializationStrategy : uint8_t { DIRECT,
INDICES };
+bool try_simd_insert_parquet_dictionary_indices(IColumn& destination, const
IColumn& dictionary,
+ const uint32_t* indices,
size_t num_values);
+
// Dictionary values are materialized once into the selected Doris type. The
state belongs to a
// column reader rather than DataTypeSerDe because a SerDe instance can be
shared by many files.
struct ParquetMaterializationState {
@@ -363,7 +366,10 @@ struct ParquetMaterializationState {
: _destination(destination), _dictionary(dictionary) {}
Status consume_indices(const uint32_t* indices, size_t num_values)
override {
- _destination.insert_indices_from(_dictionary, indices, indices
+ num_values);
+ if (!try_simd_insert_parquet_dictionary_indices(_destination,
_dictionary, indices,
+ num_values)) {
+ _destination.insert_indices_from(_dictionary, indices,
indices + num_values);
+ }
return Status::OK();
}
diff --git a/be/src/exprs/vectorized_fn_call.cpp
b/be/src/exprs/vectorized_fn_call.cpp
index 48b7ed3eaff..56ba95cd755 100644
--- a/be/src/exprs/vectorized_fn_call.cpp
+++ b/be/src/exprs/vectorized_fn_call.cpp
@@ -28,7 +28,6 @@
#include <ostream>
#include <set>
-#include "common/compare.h"
#include "common/config.h"
#include "common/exception.h"
#include "common/logging.h"
@@ -67,6 +66,7 @@
#include "storage/index/zone_map/zonemap_eval_context.h"
#include "storage/segment/column_reader.h"
#include "storage/segment/virtual_column_iterator.h"
+#include "util/simd/parquet_kernels.h"
namespace doris {
class RowDescriptor;
@@ -86,7 +86,7 @@ const static std::set<TExprOpcode::type>
OPS_FOR_ANN_RANGE_SEARCH = {
namespace {
-enum class RawComparisonOp : uint8_t { EQ, NE, LT, LE, GT, GE };
+using simd::RawComparisonOp;
std::optional<RawComparisonOp> raw_comparison_op(std::string_view
function_name, bool reverse) {
RawComparisonOp op;
@@ -128,34 +128,7 @@ template <typename T, PrimitiveType PT>
void execute_raw_comparison(const uint8_t* values, size_t num_values, const
Field& literal,
RawComparisonOp op, uint8_t* matches) {
const T rhs = literal.get<PT>();
- for (size_t row = 0; row < num_values; ++row) {
- if (matches[row] == 0) {
- continue;
- }
- const T lhs = unaligned_load<T>(values + row * sizeof(T));
- bool keep = false;
- switch (op) {
- case RawComparisonOp::EQ:
- keep = Compare::equal(lhs, rhs);
- break;
- case RawComparisonOp::NE:
- keep = Compare::not_equal(lhs, rhs);
- break;
- case RawComparisonOp::LT:
- keep = Compare::less(lhs, rhs);
- break;
- case RawComparisonOp::LE:
- keep = Compare::less_equal(lhs, rhs);
- break;
- case RawComparisonOp::GT:
- keep = Compare::greater(lhs, rhs);
- break;
- case RawComparisonOp::GE:
- keep = Compare::greater_equal(lhs, rhs);
- break;
- }
- matches[row] = static_cast<uint8_t>(keep);
- }
+ simd::raw_compare(values, num_values, rhs, op, matches);
}
} // namespace
diff --git
a/be/src/format_v2/parquet/reader/native/byte_stream_split_decoder.cpp
b/be/src/format_v2/parquet/reader/native/byte_stream_split_decoder.cpp
index caf3b675e2d..06629a43557 100644
--- a/be/src/format_v2/parquet/reader/native/byte_stream_split_decoder.cpp
+++ b/be/src/format_v2/parquet/reader/native/byte_stream_split_decoder.cpp
@@ -22,6 +22,7 @@
#include "core/column/column_fixed_length_object.h"
#include "util/byte_stream_split.h"
+#include "util/simd/parquet_kernels.h"
namespace doris::format::parquet::native {
Status ByteStreamSplitDecoder::decode_fixed_values(size_t num_values,
@@ -32,8 +33,15 @@ Status ByteStreamSplitDecoder::decode_fixed_values(size_t
num_values,
}
const int64_t stride = static_cast<int64_t>(_data->size / _type_length);
_decoded_values.resize(byte_size);
- byte_stream_split_decode(reinterpret_cast<const uint8_t*>(_data->data),
_type_length,
- _offset / _type_length, num_values, stride,
_decoded_values.data());
+ if (!simd::try_byte_stream_split_decode(reinterpret_cast<const
uint8_t*>(_data->data),
+ _type_length, _offset /
_type_length, num_values,
+ stride, _decoded_values.data())) {
+ // Unsupported widths, short batches, and non-AVX2 hosts must retain
the tuned blocked
+ // decoder instead of silently falling back to a slower row-by-row
transpose.
+ doris::byte_stream_split_decode(reinterpret_cast<const
uint8_t*>(_data->data), _type_length,
+ _offset / _type_length, num_values,
stride,
+ _decoded_values.data());
+ }
_offset += byte_size;
return consumer.consume(_decoded_values.data(), num_values,
static_cast<size_t>(_type_length));
}
@@ -57,9 +65,14 @@ Status
ByteStreamSplitDecoder::decode_selected_fixed_values(const ParquetSelecti
size_t output = 0;
const size_t first_row = _offset / value_width;
for (const auto& range : selection.ranges) {
- byte_stream_split_decode(reinterpret_cast<const
uint8_t*>(_data->data), _type_length,
- first_row + range.first, range.count, stride,
- _decoded_values.data() + output *
value_width);
+ uint8_t* destination = _decoded_values.data() + output * value_width;
+ if (!simd::try_byte_stream_split_decode(reinterpret_cast<const
uint8_t*>(_data->data),
+ _type_length, first_row +
range.first, range.count,
+ stride, destination)) {
+ doris::byte_stream_split_decode(reinterpret_cast<const
uint8_t*>(_data->data),
+ _type_length, first_row +
range.first, range.count,
+ stride, destination);
+ }
output += range.count;
}
DORIS_CHECK_EQ(output, selection.selected_values);
diff --git a/be/src/format_v2/parquet/reader/native/column_chunk_reader.cpp
b/be/src/format_v2/parquet/reader/native/column_chunk_reader.cpp
index 26afaf15bf4..4a97daaf21e 100644
--- a/be/src/format_v2/parquet/reader/native/column_chunk_reader.cpp
+++ b/be/src/format_v2/parquet/reader/native/column_chunk_reader.cpp
@@ -50,6 +50,7 @@
#include "util/bit_util.h"
#include "util/block_compression.h"
#include "util/cpu_info.h"
+#include "util/simd/parquet_kernels.h"
#include "util/unaligned.h"
namespace cctz {
@@ -603,6 +604,13 @@ void expand_nullable_pod_values(ColumnType& column, size_t
old_size, size_t comp
auto& data = column.get_data();
DORIS_CHECK_EQ(data.size(), old_size + compact_values);
data.resize(old_size + selected_nulls.size());
+ if constexpr (sizeof(typename ColumnType::value_type) == 4 ||
+ sizeof(typename ColumnType::value_type) == 8) {
+ simd::expand_nullable_values(reinterpret_cast<uint8_t*>(data.data() +
old_size),
+ compact_values, selected_nulls.data(),
selected_nulls.size(),
+ sizeof(typename ColumnType::value_type));
+ return;
+ }
size_t source = compact_values;
for (size_t output = selected_nulls.size(); output > 0;) {
--output;
diff --git a/be/src/format_v2/parquet/reader/native/delta_bit_pack_decoder.h
b/be/src/format_v2/parquet/reader/native/delta_bit_pack_decoder.h
index db9efecaced..94e248c5fe5 100644
--- a/be/src/format_v2/parquet/reader/native/delta_bit_pack_decoder.h
+++ b/be/src/format_v2/parquet/reader/native/delta_bit_pack_decoder.h
@@ -39,6 +39,7 @@
#include "format_v2/parquet/reader/native/decoder.h"
#include "util/bit_stream_utils.h"
#include "util/bit_stream_utils.inline.h"
+#include "util/simd/parquet_kernels.h"
#include "util/slice.h"
namespace doris::format::parquet::native {
@@ -787,13 +788,9 @@ Status DeltaBitPackDecoder<T>::_get_internal(T* buffer,
uint32_t num_values,
return Status::IOError("Get batch EOF");
}
}
- for (int j = 0; j < values_decode; ++j) {
- // Addition between min_delta, packed int and last_value should be
treated as
- // unsigned addition. Overflow is as expected.
- buffer[i + j] = static_cast<UT>(_min_delta) +
static_cast<UT>(buffer[i + j]) +
- static_cast<UT>(_last_value);
- _last_value = buffer[i + j];
- }
+ // Parquet defines this recurrence with wrapping integer arithmetic.
The SIMD kernel keeps
+ // the same unsigned-overflow invariant while resolving the prefix
dependency in batches.
+ simd::delta_decode(buffer + i, values_decode, _min_delta,
&_last_value);
_values_remaining_current_mini_block -= values_decode;
i += values_decode;
}
diff --git a/be/src/util/simd/parquet_kernels.cpp
b/be/src/util/simd/parquet_kernels.cpp
new file mode 100644
index 00000000000..9514ac445b0
--- /dev/null
+++ b/be/src/util/simd/parquet_kernels.cpp
@@ -0,0 +1,651 @@
+// 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 "util/simd/parquet_kernels.h"
+
+#include <array>
+#include <bit>
+#include <cmath>
+#include <cstring>
+#include <type_traits>
+
+#if defined(__x86_64__) && (defined(__GNUC__) || defined(__clang__))
+#include <immintrin.h>
+#define DORIS_PARQUET_X86_SIMD
+#endif
+
+namespace doris::simd {
+namespace {
+
+#ifdef DORIS_PARQUET_X86_SIMD
+// Keep x86-only dispatch helpers out of scalar-only builds so warning-clean
ARM builds do not
+// depend on compiler-specific unused-function behavior.
+bool has_avx2() {
+ return __builtin_cpu_supports("avx2");
+}
+
+void byte_stream_split_decode_scalar(const uint8_t* src, size_t width, size_t
offset,
+ size_t num_values, size_t stride,
uint8_t* dest) {
+ for (size_t row = 0; row < num_values; ++row) {
+ for (size_t byte = 0; byte < width; ++byte) {
+ dest[row * width + byte] = src[byte * stride + offset + row];
+ }
+ }
+}
+#endif
+
+template <typename T>
+bool scalar_compare(T lhs, T rhs, RawComparisonOp op) {
+ const auto equal = [](T left, T right) {
+ if constexpr (std::is_floating_point_v<T>) {
+ return (std::isnan(left) && std::isnan(right)) || left == right;
+ }
+ return left == right;
+ };
+ const auto greater = [](T left, T right) {
+ if constexpr (std::is_floating_point_v<T>) {
+ // Match Doris Compare: NaN is equal to NaN and greater than every
finite value.
+ if (std::isnan(right)) {
+ return false;
+ }
+ if (std::isnan(left)) {
+ return true;
+ }
+ }
+ return left > right;
+ };
+ switch (op) {
+ case RawComparisonOp::EQ:
+ return equal(lhs, rhs);
+ case RawComparisonOp::NE:
+ return !equal(lhs, rhs);
+ case RawComparisonOp::LT:
+ return greater(rhs, lhs);
+ case RawComparisonOp::LE:
+ return !greater(lhs, rhs);
+ case RawComparisonOp::GT:
+ return greater(lhs, rhs);
+ case RawComparisonOp::GE:
+ return !greater(rhs, lhs);
+ }
+ __builtin_unreachable();
+}
+
+#ifdef DORIS_PARQUET_X86_SIMD
+template <size_t WIDTH>
+__attribute__((target("avx2"))) void byte_stream_split_decode_avx2(const
uint8_t* src,
+ size_t
offset, size_t num_values,
+ size_t
stride, uint8_t* dest) {
+ static_assert(WIDTH == 4 || WIDTH == 8);
+ constexpr size_t STEPS = WIDTH == 8 ? 3 : 2;
+ constexpr size_t LANES = sizeof(__m256i);
+ const size_t blocks = num_values / LANES;
+ __m256i stage[STEPS + 1][WIDTH];
+ __m256i result[WIDTH];
+
+ for (size_t block = 0; block < blocks; ++block) {
+ for (size_t stream = 0; stream < WIDTH; ++stream) {
+ stage[0][stream] = _mm256_loadu_si256(reinterpret_cast<const
__m256i*>(
+ src + stream * stride + offset + block * LANES));
+ }
+ for (size_t step = 0; step < STEPS; ++step) {
+ // AVX2 unpack instructions operate independently in each 128-bit
half. Keep the
+ // byte-transpose hierarchy lane-local, then stitch the halves
only after the last
+ // unpack so every output vector contains contiguous decoded rows.
+ for (size_t pair = 0; pair < WIDTH / 2; ++pair) {
+ stage[step + 1][pair * 2] =
+ _mm256_unpacklo_epi8(stage[step][pair],
stage[step][WIDTH / 2 + pair]);
+ stage[step + 1][pair * 2 + 1] =
+ _mm256_unpackhi_epi8(stage[step][pair],
stage[step][WIDTH / 2 + pair]);
+ }
+ }
+ for (size_t pair = 0; pair < WIDTH / 2; ++pair) {
+ result[pair] = _mm256_permute2x128_si256(stage[STEPS][pair * 2],
+ stage[STEPS][pair * 2 +
1], 0x20);
+ result[WIDTH / 2 + pair] =
_mm256_permute2x128_si256(stage[STEPS][pair * 2],
+
stage[STEPS][pair * 2 + 1], 0x31);
+ }
+ for (size_t lane = 0; lane < WIDTH; ++lane) {
+ _mm256_storeu_si256(reinterpret_cast<__m256i*>(dest + (block *
WIDTH + lane) * LANES),
+ result[lane]);
+ }
+ }
+ const size_t processed = blocks * LANES;
+ byte_stream_split_decode_scalar(src, WIDTH, offset + processed, num_values
- processed, stride,
+ dest + processed * WIDTH);
+}
+
+__attribute__((target("avx2"))) void delta_decode_int32_avx2(int32_t* values,
size_t count,
+ int32_t min_delta,
+ int32_t*
last_value) {
+ const __m256i min_delta_vec = _mm256_set1_epi32(min_delta);
+ const size_t vector_count = count / 8 * 8;
+ for (size_t row = 0; row < vector_count; row += 8) {
+ __m256i value = _mm256_loadu_si256(reinterpret_cast<const
__m256i*>(values + row));
+ value = _mm256_add_epi32(value, min_delta_vec);
+ value = _mm256_add_epi32(value, _mm256_slli_si256(value, 4));
+ value = _mm256_add_epi32(value, _mm256_slli_si256(value, 8));
+ _mm256_storeu_si256(reinterpret_cast<__m256i*>(values + row), value);
+ }
+ __m128i carry = _mm_set1_epi32(*last_value);
+ for (size_t row = 0; row < vector_count; row += 4) {
+ __m128i value = _mm_loadu_si128(reinterpret_cast<const
__m128i*>(values + row));
+ value = _mm_add_epi32(value, carry);
+ _mm_storeu_si128(reinterpret_cast<__m128i*>(values + row), value);
+ carry = _mm_shuffle_epi32(value, _MM_SHUFFLE(3, 3, 3, 3));
+ }
+ if (vector_count != 0) {
+ *last_value = _mm_cvtsi128_si32(carry);
+ }
+ using Unsigned = uint32_t;
+ for (size_t row = vector_count; row < count; ++row) {
+ values[row] = static_cast<int32_t>(static_cast<Unsigned>(values[row]) +
+ static_cast<Unsigned>(min_delta) +
+ static_cast<Unsigned>(*last_value));
+ *last_value = values[row];
+ }
+}
+
+__attribute__((target("avx2"))) void delta_decode_int64_avx2(int64_t* values,
size_t count,
+ int64_t min_delta,
+ int64_t*
last_value) {
+ const __m256i min_delta_vec = _mm256_set1_epi64x(min_delta);
+ const size_t vector_count = count / 4 * 4;
+ for (size_t row = 0; row < vector_count; row += 4) {
+ __m256i value = _mm256_loadu_si256(reinterpret_cast<const
__m256i*>(values + row));
+ value = _mm256_add_epi64(value, min_delta_vec);
+ value = _mm256_add_epi64(value, _mm256_slli_si256(value, 8));
+ _mm256_storeu_si256(reinterpret_cast<__m256i*>(values + row), value);
+ }
+ __m128i carry = _mm_set1_epi64x(*last_value);
+ for (size_t row = 0; row < vector_count; row += 2) {
+ __m128i value = _mm_loadu_si128(reinterpret_cast<const
__m128i*>(values + row));
+ value = _mm_add_epi64(value, carry);
+ _mm_storeu_si128(reinterpret_cast<__m128i*>(values + row), value);
+ carry = _mm_unpackhi_epi64(value, value);
+ }
+ if (vector_count != 0) {
+ *last_value = _mm_cvtsi128_si64(carry);
+ }
+ using Unsigned = uint64_t;
+ for (size_t row = vector_count; row < count; ++row) {
+ values[row] = static_cast<int64_t>(static_cast<Unsigned>(values[row]) +
+ static_cast<Unsigned>(min_delta) +
+ static_cast<Unsigned>(*last_value));
+ *last_value = values[row];
+ }
+}
+
+__attribute__((target("avx2"))) void dictionary_gather_avx2(const uint8_t*
dictionary,
+ const uint32_t*
indices, size_t count,
+ size_t
value_width, uint8_t* dest) {
+ size_t row = 0;
+ if (value_width == 4) {
+ for (; row + 8 <= count; row += 8) {
+ const __m256i ids = _mm256_loadu_si256(reinterpret_cast<const
__m256i*>(indices + row));
+ const __m256i gathered =
+ _mm256_i32gather_epi32(reinterpret_cast<const
int*>(dictionary), ids, 4);
+ _mm256_storeu_si256(reinterpret_cast<__m256i*>(dest + row * 4),
gathered);
+ }
+ } else {
+ for (; row + 4 <= count; row += 4) {
+ const __m128i ids = _mm_loadu_si128(reinterpret_cast<const
__m128i*>(indices + row));
+ const __m256i gathered =
+ _mm256_i32gather_epi64(reinterpret_cast<const long
long*>(dictionary), ids, 8);
+ _mm256_storeu_si256(reinterpret_cast<__m256i*>(dest + row * 8),
gathered);
+ }
+ }
+ for (; row < count; ++row) {
+ memcpy(dest + row * value_width, dictionary + indices[row] *
value_width, value_width);
+ }
+}
+
+template <size_t LANES>
+constexpr auto make_expand_permute_lut() {
+ std::array<std::array<int32_t, 8>, 1U << LANES> lut {};
+ for (size_t mask = 0; mask < lut.size(); ++mask) {
+ int32_t source = 0;
+ for (size_t lane = 0; lane < LANES; ++lane) {
+ const int32_t value = (mask & (1U << lane)) != 0 ? source++ : 0;
+ if constexpr (LANES == 8) {
+ lut[mask][lane] = value;
+ } else {
+ lut[mask][lane * 2] = value * 2;
+ lut[mask][lane * 2 + 1] = value * 2 + 1;
+ }
+ }
+ }
+ return lut;
+}
+
+constexpr auto EXPAND_PERMUTE_32 = make_expand_permute_lut<8>();
+constexpr auto EXPAND_PERMUTE_64 = make_expand_permute_lut<4>();
+
+__attribute__((target("avx2"))) void expand_nullable_avx2(uint8_t* bytes,
size_t compact_count,
+ const uint8_t*
nulls, size_t output_count,
+ size_t value_width) {
+ size_t source = compact_count;
+ size_t output = output_count;
+ if (value_width == 4) {
+ auto* values = reinterpret_cast<int32_t*>(bytes);
+ while (output >= 8) {
+ const size_t start = output - 8;
+ uint32_t valid_mask = 0;
+ for (size_t lane = 0; lane < 8; ++lane) {
+ valid_mask |= static_cast<uint32_t>(nulls[start + lane] == 0)
<< lane;
+ }
+ const size_t valid = std::popcount(valid_mask);
+ source -= valid;
+ const __m256i load_mask =
_mm256_cmpgt_epi32(_mm256_set1_epi32(static_cast<int>(valid)),
+ _mm256_setr_epi32(0,
1, 2, 3, 4, 5, 6, 7));
+ const __m256i compact = _mm256_maskload_epi32(values + source,
load_mask);
+ const __m256i permute = _mm256_loadu_si256(
+ reinterpret_cast<const
__m256i*>(EXPAND_PERMUTE_32[valid_mask].data()));
+ __m256i expanded = _mm256_permutevar8x32_epi32(compact, permute);
+ const __m256i valid_lanes = _mm256_setr_epi32(
+ -(valid_mask & 1U), -((valid_mask >> 1) & 1U),
-((valid_mask >> 2) & 1U),
+ -((valid_mask >> 3) & 1U), -((valid_mask >> 4) & 1U),
-((valid_mask >> 5) & 1U),
+ -((valid_mask >> 6) & 1U), -((valid_mask >> 7) & 1U));
+ expanded = _mm256_and_si256(expanded, valid_lanes);
+ _mm256_storeu_si256(reinterpret_cast<__m256i*>(values + start),
expanded);
+ output = start;
+ }
+ } else {
+ auto* values = reinterpret_cast<int64_t*>(bytes);
+ while (output >= 4) {
+ const size_t start = output - 4;
+ uint32_t valid_mask = 0;
+ for (size_t lane = 0; lane < 4; ++lane) {
+ valid_mask |= static_cast<uint32_t>(nulls[start + lane] == 0)
<< lane;
+ }
+ const size_t valid = std::popcount(valid_mask);
+ source -= valid;
+ const __m256i load_mask =
+ _mm256_setr_epi64x(valid > 0 ? -1LL : 0, valid > 1 ? -1LL
: 0,
+ valid > 2 ? -1LL : 0, valid > 3 ? -1LL
: 0);
+ const __m256i compact = _mm256_maskload_epi64(
+ reinterpret_cast<const long long*>(values + source),
load_mask);
+ const __m256i permute = _mm256_loadu_si256(
+ reinterpret_cast<const
__m256i*>(EXPAND_PERMUTE_64[valid_mask].data()));
+ __m256i expanded = _mm256_permutevar8x32_epi32(compact, permute);
+ const __m256i valid_lanes = _mm256_setr_epi64x(
+ (valid_mask & 1U) != 0 ? -1LL : 0, (valid_mask & 2U) != 0
? -1LL : 0,
+ (valid_mask & 4U) != 0 ? -1LL : 0, (valid_mask & 8U) != 0
? -1LL : 0);
+ expanded = _mm256_and_si256(expanded, valid_lanes);
+ _mm256_storeu_si256(reinterpret_cast<__m256i*>(values + start),
expanded);
+ output = start;
+ }
+ }
+ while (output > 0) {
+ --output;
+ if (nulls[output] != 0) {
+ memset(bytes + output * value_width, 0, value_width);
+ } else {
+ --source;
+ memmove(bytes + output * value_width, bytes + source *
value_width, value_width);
+ }
+ }
+}
+
+__attribute__((target("avx2"))) __m256i vector_all(__m256i) {
+ return _mm256_set1_epi32(-1);
+}
+
+__attribute__((target("avx2"))) __m256 vector_all(__m256) {
+ return _mm256_castsi256_ps(_mm256_set1_epi32(-1));
+}
+
+__attribute__((target("avx2"))) __m256d vector_all(__m256d) {
+ return _mm256_castsi256_pd(_mm256_set1_epi32(-1));
+}
+
+__attribute__((target("avx2"))) __m256i vector_or(__m256i lhs, __m256i rhs) {
+ return _mm256_or_si256(lhs, rhs);
+}
+
+__attribute__((target("avx2"))) __m256 vector_or(__m256 lhs, __m256 rhs) {
+ return _mm256_or_ps(lhs, rhs);
+}
+
+__attribute__((target("avx2"))) __m256d vector_or(__m256d lhs, __m256d rhs) {
+ return _mm256_or_pd(lhs, rhs);
+}
+
+__attribute__((target("avx2"))) __m256i vector_xor(__m256i lhs, __m256i rhs) {
+ return _mm256_xor_si256(lhs, rhs);
+}
+
+__attribute__((target("avx2"))) __m256 vector_xor(__m256 lhs, __m256 rhs) {
+ return _mm256_xor_ps(lhs, rhs);
+}
+
+__attribute__((target("avx2"))) __m256d vector_xor(__m256d lhs, __m256d rhs) {
+ return _mm256_xor_pd(lhs, rhs);
+}
+
+template <typename Vec>
+__attribute__((target("avx2"))) Vec combine_comparison(Vec equal, Vec greater,
Vec less,
+ RawComparisonOp op) {
+ // Target features do not propagate from AVX2 callers into separately
instantiated helpers.
+ // Keep the complete vector operation inside its own target scope for
baseline x86 builds.
+ const Vec all = vector_all(equal);
+ switch (op) {
+ case RawComparisonOp::EQ:
+ return equal;
+ case RawComparisonOp::NE:
+ return vector_xor(equal, all);
+ case RawComparisonOp::LT:
+ return less;
+ case RawComparisonOp::LE:
+ return vector_or(less, equal);
+ case RawComparisonOp::GT:
+ return greater;
+ case RawComparisonOp::GE:
+ return vector_or(greater, equal);
+ }
+ __builtin_unreachable();
+}
+
+__attribute__((target("avx2"))) void raw_compare_int32_avx2(const uint8_t*
bytes, size_t count,
+ int32_t literal,
RawComparisonOp op,
+ uint8_t* matches) {
+ size_t row = 0;
+ const __m256i rhs = _mm256_set1_epi32(literal);
+ for (; row + 8 <= count; row += 8) {
+ const __m256i lhs = _mm256_loadu_si256(reinterpret_cast<const
__m256i*>(bytes + row * 4));
+ const __m256i equal = _mm256_cmpeq_epi32(lhs, rhs);
+ const __m256i greater = _mm256_cmpgt_epi32(lhs, rhs);
+ const __m256i less = _mm256_cmpgt_epi32(rhs, lhs);
+ const uint32_t mask = static_cast<uint32_t>(_mm256_movemask_ps(
+ _mm256_castsi256_ps(combine_comparison(equal, greater, less,
op))));
+ for (size_t lane = 0; lane < 8; ++lane) {
+ matches[row + lane] &= static_cast<uint8_t>((mask >> lane) & 1U);
+ }
+ }
+ for (; row < count; ++row) {
+ int32_t value;
+ memcpy(&value, bytes + row * 4, sizeof(value));
+ bool keep = false;
+ switch (op) {
+ case RawComparisonOp::EQ:
+ keep = value == literal;
+ break;
+ case RawComparisonOp::NE:
+ keep = value != literal;
+ break;
+ case RawComparisonOp::LT:
+ keep = value < literal;
+ break;
+ case RawComparisonOp::LE:
+ keep = value <= literal;
+ break;
+ case RawComparisonOp::GT:
+ keep = value > literal;
+ break;
+ case RawComparisonOp::GE:
+ keep = value >= literal;
+ break;
+ }
+ matches[row] &= static_cast<uint8_t>(keep);
+ }
+}
+
+__attribute__((target("avx2"))) void raw_compare_int64_avx2(const uint8_t*
bytes, size_t count,
+ int64_t literal,
RawComparisonOp op,
+ uint8_t* matches) {
+ size_t row = 0;
+ const __m256i rhs = _mm256_set1_epi64x(literal);
+ for (; row + 4 <= count; row += 4) {
+ const __m256i lhs = _mm256_loadu_si256(reinterpret_cast<const
__m256i*>(bytes + row * 8));
+ const __m256i equal = _mm256_cmpeq_epi64(lhs, rhs);
+ const __m256i greater = _mm256_cmpgt_epi64(lhs, rhs);
+ const __m256i less = _mm256_cmpgt_epi64(rhs, lhs);
+ const uint32_t mask = static_cast<uint32_t>(_mm256_movemask_pd(
+ _mm256_castsi256_pd(combine_comparison(equal, greater, less,
op))));
+ for (size_t lane = 0; lane < 4; ++lane) {
+ matches[row + lane] &= static_cast<uint8_t>((mask >> lane) & 1U);
+ }
+ }
+ for (; row < count; ++row) {
+ int64_t value;
+ memcpy(&value, bytes + row * 8, sizeof(value));
+ bool keep = false;
+ switch (op) {
+ case RawComparisonOp::EQ:
+ keep = value == literal;
+ break;
+ case RawComparisonOp::NE:
+ keep = value != literal;
+ break;
+ case RawComparisonOp::LT:
+ keep = value < literal;
+ break;
+ case RawComparisonOp::LE:
+ keep = value <= literal;
+ break;
+ case RawComparisonOp::GT:
+ keep = value > literal;
+ break;
+ case RawComparisonOp::GE:
+ keep = value >= literal;
+ break;
+ }
+ matches[row] &= static_cast<uint8_t>(keep);
+ }
+}
+
+__attribute__((target("avx2"))) void raw_compare_float_avx2(const uint8_t*
bytes, size_t count,
+ float literal,
RawComparisonOp op,
+ uint8_t* matches) {
+ size_t row = 0;
+ const __m256 rhs = _mm256_set1_ps(literal);
+ for (; row + 8 <= count; row += 8) {
+ const __m256 lhs = _mm256_loadu_ps(reinterpret_cast<const
float*>(bytes + row * 4));
+ const __m256 lhs_nan = _mm256_cmp_ps(lhs, lhs, _CMP_UNORD_Q);
+ const __m256 rhs_nan = _mm256_cmp_ps(rhs, rhs, _CMP_UNORD_Q);
+ const __m256 both_nan = _mm256_and_ps(lhs_nan, rhs_nan);
+ const __m256 equal = _mm256_or_ps(_mm256_cmp_ps(lhs, rhs, _CMP_EQ_OQ),
both_nan);
+ const __m256 greater = _mm256_or_ps(_mm256_cmp_ps(lhs, rhs,
_CMP_GT_OQ),
+ _mm256_andnot_ps(rhs_nan,
lhs_nan));
+ const __m256 less = _mm256_or_ps(_mm256_cmp_ps(lhs, rhs, _CMP_LT_OQ),
+ _mm256_andnot_ps(lhs_nan, rhs_nan));
+ const uint32_t mask = static_cast<uint32_t>(
+ _mm256_movemask_ps(combine_comparison(equal, greater, less,
op)));
+ for (size_t lane = 0; lane < 8; ++lane) {
+ matches[row + lane] &= static_cast<uint8_t>((mask >> lane) & 1U);
+ }
+ }
+ for (; row < count; ++row) {
+ float value;
+ memcpy(&value, bytes + row * 4, sizeof(value));
+ matches[row] &= static_cast<uint8_t>(scalar_compare(value, literal,
op));
+ }
+}
+
+__attribute__((target("avx2"))) void raw_compare_double_avx2(const uint8_t*
bytes, size_t count,
+ double literal,
RawComparisonOp op,
+ uint8_t* matches)
{
+ size_t row = 0;
+ const __m256d rhs = _mm256_set1_pd(literal);
+ for (; row + 4 <= count; row += 4) {
+ const __m256d lhs = _mm256_loadu_pd(reinterpret_cast<const
double*>(bytes + row * 8));
+ const __m256d lhs_nan = _mm256_cmp_pd(lhs, lhs, _CMP_UNORD_Q);
+ const __m256d rhs_nan = _mm256_cmp_pd(rhs, rhs, _CMP_UNORD_Q);
+ const __m256d both_nan = _mm256_and_pd(lhs_nan, rhs_nan);
+ const __m256d equal = _mm256_or_pd(_mm256_cmp_pd(lhs, rhs,
_CMP_EQ_OQ), both_nan);
+ const __m256d greater = _mm256_or_pd(_mm256_cmp_pd(lhs, rhs,
_CMP_GT_OQ),
+ _mm256_andnot_pd(rhs_nan,
lhs_nan));
+ const __m256d less = _mm256_or_pd(_mm256_cmp_pd(lhs, rhs, _CMP_LT_OQ),
+ _mm256_andnot_pd(lhs_nan, rhs_nan));
+ const uint32_t mask = static_cast<uint32_t>(
+ _mm256_movemask_pd(combine_comparison(equal, greater, less,
op)));
+ for (size_t lane = 0; lane < 4; ++lane) {
+ matches[row + lane] &= static_cast<uint8_t>((mask >> lane) & 1U);
+ }
+ }
+ for (; row < count; ++row) {
+ double value;
+ memcpy(&value, bytes + row * 8, sizeof(value));
+ matches[row] &= static_cast<uint8_t>(scalar_compare(value, literal,
op));
+ }
+}
+#endif
+
+template <typename T>
+void delta_decode_scalar(T* values, size_t count, T min_delta, T* last_value) {
+ using Unsigned = std::make_unsigned_t<T>;
+ for (size_t row = 0; row < count; ++row) {
+ values[row] = static_cast<T>(static_cast<Unsigned>(values[row]) +
+ static_cast<Unsigned>(min_delta) +
+ static_cast<Unsigned>(*last_value));
+ *last_value = values[row];
+ }
+}
+
+template <typename T>
+void raw_compare_scalar(const uint8_t* bytes, size_t count, T literal,
RawComparisonOp op,
+ uint8_t* matches) {
+ for (size_t row = 0; row < count; ++row) {
+ if (matches[row] == 0) {
+ continue;
+ }
+ T value;
+ memcpy(&value, bytes + row * sizeof(T), sizeof(T));
+ matches[row] = static_cast<uint8_t>(scalar_compare(value, literal,
op));
+ }
+}
+
+} // namespace
+
+bool try_byte_stream_split_decode(const uint8_t* src, size_t width, size_t
offset,
+ size_t num_values, size_t stride, uint8_t*
dest) {
+#ifdef DORIS_PARQUET_X86_SIMD
+ if (has_avx2() && num_values >= 32 && (width == 4 || width == 8)) {
+ if (width == 4) {
+ byte_stream_split_decode_avx2<4>(src, offset, num_values, stride,
dest);
+ } else {
+ byte_stream_split_decode_avx2<8>(src, offset, num_values, stride,
dest);
+ }
+ return true;
+ }
+#endif
+ return false;
+}
+
+void delta_decode(int32_t* values, size_t count, int32_t min_delta, int32_t*
last_value) {
+#ifdef DORIS_PARQUET_X86_SIMD
+ if (has_avx2() && count >= 8) {
+ delta_decode_int32_avx2(values, count, min_delta, last_value);
+ return;
+ }
+#endif
+ delta_decode_scalar(values, count, min_delta, last_value);
+}
+
+void delta_decode(int64_t* values, size_t count, int64_t min_delta, int64_t*
last_value) {
+#ifdef DORIS_PARQUET_X86_SIMD
+ if (has_avx2() && count >= 4) {
+ delta_decode_int64_avx2(values, count, min_delta, last_value);
+ return;
+ }
+#endif
+ delta_decode_scalar(values, count, min_delta, last_value);
+}
+
+void dictionary_gather(const uint8_t* dictionary, const uint32_t* indices,
size_t count,
+ size_t value_width, uint8_t* dest) {
+#ifdef DORIS_PARQUET_X86_SIMD
+ if (has_avx2() && ((value_width == 4 && count >= 8) || (value_width == 8
&& count >= 4))) {
+ dictionary_gather_avx2(dictionary, indices, count, value_width, dest);
+ return;
+ }
+#endif
+ for (size_t row = 0; row < count; ++row) {
+ memcpy(dest + row * value_width, dictionary + indices[row] *
value_width, value_width);
+ }
+}
+
+void expand_nullable_values(uint8_t* values, size_t compact_count, const
uint8_t* nulls,
+ size_t output_count, size_t value_width) {
+#ifdef DORIS_PARQUET_X86_SIMD
+ if (has_avx2() &&
+ ((value_width == 4 && output_count >= 8) || (value_width == 8 &&
output_count >= 4))) {
+ // Backward expansion is required because the compact input and
expanded output alias.
+ // Each SIMD block loads all of its source lanes before overwriting
the wider destination.
+ expand_nullable_avx2(values, compact_count, nulls, output_count,
value_width);
+ return;
+ }
+#endif
+ size_t source = compact_count;
+ for (size_t output = output_count; output > 0;) {
+ --output;
+ if (nulls[output] != 0) {
+ memset(values + output * value_width, 0, value_width);
+ } else {
+ --source;
+ memmove(values + output * value_width, values + source *
value_width, value_width);
+ }
+ }
+}
+
+void raw_compare(const uint8_t* values, size_t count, int32_t literal,
RawComparisonOp op,
+ uint8_t* matches) {
+#ifdef DORIS_PARQUET_X86_SIMD
+ if (has_avx2() && count >= 8) {
+ raw_compare_int32_avx2(values, count, literal, op, matches);
+ return;
+ }
+#endif
+ raw_compare_scalar(values, count, literal, op, matches);
+}
+
+void raw_compare(const uint8_t* values, size_t count, int64_t literal,
RawComparisonOp op,
+ uint8_t* matches) {
+#ifdef DORIS_PARQUET_X86_SIMD
+ if (has_avx2() && count >= 4) {
+ raw_compare_int64_avx2(values, count, literal, op, matches);
+ return;
+ }
+#endif
+ raw_compare_scalar(values, count, literal, op, matches);
+}
+
+void raw_compare(const uint8_t* values, size_t count, float literal,
RawComparisonOp op,
+ uint8_t* matches) {
+#ifdef DORIS_PARQUET_X86_SIMD
+ if (has_avx2() && count >= 8) {
+ // Doris orders NaN above every finite value and considers NaN equal
to NaN. The SIMD
+ // masks deliberately reconstruct that total order instead of using
ordered FP compares.
+ raw_compare_float_avx2(values, count, literal, op, matches);
+ return;
+ }
+#endif
+ raw_compare_scalar(values, count, literal, op, matches);
+}
+
+void raw_compare(const uint8_t* values, size_t count, double literal,
RawComparisonOp op,
+ uint8_t* matches) {
+#ifdef DORIS_PARQUET_X86_SIMD
+ if (has_avx2() && count >= 4) {
+ raw_compare_double_avx2(values, count, literal, op, matches);
+ return;
+ }
+#endif
+ raw_compare_scalar(values, count, literal, op, matches);
+}
+
+} // namespace doris::simd
diff --git a/be/src/util/simd/parquet_kernels.h
b/be/src/util/simd/parquet_kernels.h
new file mode 100644
index 00000000000..2381b85e41c
--- /dev/null
+++ b/be/src/util/simd/parquet_kernels.h
@@ -0,0 +1,48 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#pragma once
+
+#include <cstddef>
+#include <cstdint>
+
+namespace doris::simd {
+
+enum class RawComparisonOp : uint8_t { EQ, NE, LT, LE, GT, GE };
+
+bool try_byte_stream_split_decode(const uint8_t* src, size_t width, size_t
offset,
+ size_t num_values, size_t stride, uint8_t*
dest);
+
+void delta_decode(int32_t* values, size_t count, int32_t min_delta, int32_t*
last_value);
+void delta_decode(int64_t* values, size_t count, int64_t min_delta, int64_t*
last_value);
+
+void dictionary_gather(const uint8_t* dictionary, const uint32_t* indices,
size_t count,
+ size_t value_width, uint8_t* dest);
+
+void expand_nullable_values(uint8_t* values, size_t compact_count, const
uint8_t* nulls,
+ size_t output_count, size_t value_width);
+
+void raw_compare(const uint8_t* values, size_t count, int32_t literal,
RawComparisonOp op,
+ uint8_t* matches);
+void raw_compare(const uint8_t* values, size_t count, int64_t literal,
RawComparisonOp op,
+ uint8_t* matches);
+void raw_compare(const uint8_t* values, size_t count, float literal,
RawComparisonOp op,
+ uint8_t* matches);
+void raw_compare(const uint8_t* values, size_t count, double literal,
RawComparisonOp op,
+ uint8_t* matches);
+
+} // namespace doris::simd
diff --git a/be/test/format_v2/parquet/native_decoder_test.cpp
b/be/test/format_v2/parquet/native_decoder_test.cpp
index 599c930f312..9192a055f01 100644
--- a/be/test/format_v2/parquet/native_decoder_test.cpp
+++ b/be/test/format_v2/parquet/native_decoder_test.cpp
@@ -1269,12 +1269,14 @@ TEST(ParquetV2NativeDecoderTest,
DictionaryMaterializationUsesCacheAwareExecutio
ParquetMaterializationState state;
state.typed_dictionary = ColumnInt32::create();
assert_cast<ColumnInt32&>(*state.typed_dictionary).get_data() = {10,
20, 30, 40};
- ScriptedDictionaryMaterializationSource source({3, 0, 2, 1, 3},
prefer_indices);
+ const std::vector<uint32_t> ids {3, 0, 2, 1, 3, 1, 0, 2, 3, 3, 1};
+ ScriptedDictionaryMaterializationSource source(ids, prefer_indices);
auto output = ColumnInt32::create();
- const auto status = state.materialize_dictionary(*output, source, 5);
+ const auto status = state.materialize_dictionary(*output, source,
ids.size());
EXPECT_TRUE(status.ok()) << status;
- EXPECT_EQ(output->get_data(), (ColumnInt32::Container {40, 10, 30, 20,
40}));
+ EXPECT_EQ(output->get_data(),
+ (ColumnInt32::Container {40, 10, 30, 20, 40, 20, 10, 30, 40,
40, 20}));
EXPECT_EQ(state.dictionary_materialization_strategy,
expected_strategy);
EXPECT_EQ(source.direct_batches, expected_direct_batches);
EXPECT_EQ(source.index_batches, expected_index_batches);
diff --git a/be/test/format_v2/parquet/parquet_benchmark_scenarios_test.cpp
b/be/test/format_v2/parquet/parquet_benchmark_scenarios_test.cpp
index 45c1fe9dd44..7cc0838e125 100644
--- a/be/test/format_v2/parquet/parquet_benchmark_scenarios_test.cpp
+++ b/be/test/format_v2/parquet/parquet_benchmark_scenarios_test.cpp
@@ -21,6 +21,7 @@
#include <algorithm>
#include <array>
+#include <map>
#include <set>
#include <string>
#include <tuple>
@@ -30,6 +31,7 @@ namespace {
TEST(ParquetBenchmarkScenariosTest,
DecoderMatrixCoversNativeEncodingAndTypeFamilies) {
const auto scenarios = decoder_scenarios();
+ EXPECT_EQ(scenarios.size() * 6 * 2, size_t {228});
const std::set<std::pair<Encoding, ValueType>> actual = [&] {
std::set<std::pair<Encoding, ValueType>> values;
for (const auto& scenario : scenarios) {
@@ -62,8 +64,55 @@ TEST(ParquetBenchmarkScenariosTest,
DecoderMatrixCoversNativeEncodingAndTypeFami
EXPECT_EQ(actual, expected);
}
+TEST(ParquetBenchmarkScenariosTest,
KernelMatrixCoversEverySimdStageAndBoundaryShape) {
+ const auto scenarios = kernel_scenarios();
+ EXPECT_EQ(scenarios.size(), size_t {80});
+ const std::map<Kernel, std::vector<ValueType>> expected_types {
+ {Kernel::BYTE_STREAM_SPLIT, {ValueType::FLOAT, ValueType::DOUBLE}},
+ {Kernel::DELTA_PREFIX_SUM, {ValueType::INT32, ValueType::INT64}},
+ {Kernel::DICTIONARY_GATHER,
+ {ValueType::INT32, ValueType::INT64, ValueType::FLOAT,
ValueType::DOUBLE}},
+ {Kernel::NULLABLE_EXPAND,
+ {ValueType::INT32, ValueType::INT64, ValueType::FLOAT,
ValueType::DOUBLE}},
+ {Kernel::RAW_PREDICATE,
+ {ValueType::INT32, ValueType::INT64, ValueType::FLOAT,
ValueType::DOUBLE}},
+ };
+ for (const auto& [kernel, value_types] : expected_types) {
+ for (const auto value_type : value_types) {
+ EXPECT_TRUE(std::ranges::any_of(scenarios,
+ [&](const KernelScenario&
scenario) {
+ return scenario.kernel ==
kernel &&
+ scenario.value_type ==
value_type;
+ }))
+ << "missing SIMD width for kernel " << to_string(kernel);
+ }
+ }
+ for (const int selectivity : {0, 1, 10, 50, 90, 100}) {
+ EXPECT_TRUE(std::ranges::any_of(scenarios, [&](const KernelScenario&
scenario) {
+ return scenario.kernel == Kernel::RAW_PREDICATE &&
+ scenario.selectivity_percent == selectivity;
+ })) << "missing raw predicate selectivity";
+ }
+ for (const int null_percent : {0, 1, 10, 50, 90}) {
+ for (const auto pattern : {Pattern::CLUSTERED, Pattern::ALTERNATING}) {
+ EXPECT_TRUE(std::ranges::any_of(scenarios, [&](const
KernelScenario& scenario) {
+ return scenario.kernel == Kernel::NULLABLE_EXPAND &&
+ scenario.null_percent == null_percent &&
scenario.pattern == pattern;
+ })) << "missing nullable expansion shape";
+ }
+ }
+ for (const size_t dictionary_entries : {32, 4096, 262144}) {
+ EXPECT_TRUE(std::ranges::any_of(scenarios, [&](const KernelScenario&
scenario) {
+ return scenario.kernel == Kernel::DICTIONARY_GATHER &&
+ scenario.dictionary_entries == dictionary_entries;
+ })) << "missing dictionary working-set size";
+ }
+}
+
TEST(ParquetBenchmarkScenariosTest,
ReaderMatrixCoversNullableSparseAndProjectionAxes) {
const auto scenarios = reader_scenarios();
+ // Keep the exact count aligned with the upstream complex-residual
scenario retained by rebase.
+ EXPECT_EQ(scenarios.size(), size_t {152});
for (const int null_percent : {0, 1, 10, 50, 90}) {
for (const auto pattern : {Pattern::CLUSTERED, Pattern::ALTERNATING}) {
for (const int selectivity : {0, 1, 10, 50, 90, 100}) {
@@ -113,7 +162,7 @@ TEST(ParquetBenchmarkScenariosTest,
ReaderMatrixCoversOperationsEncodingsAndSche
TEST(ParquetBenchmarkScenariosTest,
ReaderMatrixHasExactUniqueRegistrationNames) {
const auto scenarios = reader_scenarios();
- EXPECT_EQ(scenarios.size(), 152);
+ EXPECT_EQ(scenarios.size(), size_t {152});
std::set<std::string> names;
for (const auto& scenario : scenarios) {
@@ -166,5 +215,18 @@ TEST(ParquetBenchmarkScenariosTest,
SelectionPlanDistinguishesClusteredAndSparse
EXPECT_EQ(make_selection_plan(1000, 100,
Pattern::ALTERNATING).ranges.size(), 1);
}
+TEST(ParquetBenchmarkScenariosTest,
SelectedRowVisitorPreservesRangeOrderAndBoundaries) {
+ const SelectionPlan plan {
+ .total_rows = 12,
+ .selected_rows = 5,
+ .ranges = {{.first = 0, .count = 2},
+ {.first = 5, .count = 1},
+ {.first = 9, .count = 2}},
+ };
+ std::vector<size_t> rows;
+ visit_selected_rows(plan, [&](size_t row) { rows.push_back(row); });
+ EXPECT_EQ(rows, (std::vector<size_t> {0, 1, 5, 9, 10}));
+}
+
} // namespace
} // namespace doris::parquet_benchmark
diff --git a/be/test/format_v2/parquet/parquet_simd_kernels_test.cpp
b/be/test/format_v2/parquet/parquet_simd_kernels_test.cpp
new file mode 100644
index 00000000000..c1a674f7aaa
--- /dev/null
+++ b/be/test/format_v2/parquet/parquet_simd_kernels_test.cpp
@@ -0,0 +1,222 @@
+// 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 <gtest/gtest.h>
+
+#include <algorithm>
+#include <array>
+#include <cmath>
+#include <cstdint>
+#include <cstring>
+#include <limits>
+#include <type_traits>
+#include <vector>
+
+#include "util/byte_stream_split.h"
+#include "util/simd/parquet_kernels.h"
+
+namespace doris::simd {
+namespace {
+
+TEST(ParquetSimdKernelsTest, ByteStreamSplitRestoresFourAndEightByteValues) {
+ for (const size_t width : {4, 8}) {
+ constexpr size_t rows = 67;
+ std::vector<uint8_t> plain(rows * width);
+ for (size_t byte = 0; byte < plain.size(); ++byte) {
+ plain[byte] = static_cast<uint8_t>((byte * 29 + 17) & 0xff);
+ }
+ std::vector<uint8_t> encoded(plain.size());
+ for (size_t row = 0; row < rows; ++row) {
+ for (size_t byte = 0; byte < width; ++byte) {
+ encoded[byte * rows + row] = plain[row * width + byte];
+ }
+ }
+
+ for (const auto [offset, count] :
+ {std::pair<size_t, size_t> {0, rows}, {3, 31}, {17, 33}}) {
+ std::vector<uint8_t> decoded(count * width);
+ if (!try_byte_stream_split_decode(encoded.data(), width, offset,
count, rows,
+ decoded.data())) {
+ doris::byte_stream_split_decode(encoded.data(),
static_cast<int>(width), offset,
+ count, rows, decoded.data());
+ }
+ EXPECT_EQ(0, memcmp(decoded.data(), plain.data() + offset * width,
decoded.size()));
+ }
+ }
+}
+
+TEST(ParquetSimdKernelsTest,
ByteStreamSplitDeclinesToReplaceOptimizedScalarFallback) {
+ constexpr size_t rows = 191;
+ constexpr size_t offset = 17;
+ constexpr size_t count = 31;
+ for (const size_t width : {1, 2, 4, 8, 12, 16, 23}) {
+ std::vector<uint8_t> plain(rows * width);
+ for (size_t byte = 0; byte < plain.size(); ++byte) {
+ plain[byte] = static_cast<uint8_t>((byte * 29 + 17) & 0xff);
+ }
+ std::vector<uint8_t> encoded(plain.size());
+ for (size_t row = 0; row < rows; ++row) {
+ for (size_t byte = 0; byte < width; ++byte) {
+ encoded[byte * rows + row] = plain[row * width + byte];
+ }
+ }
+
+ std::vector<uint8_t> decoded(count * width);
+ EXPECT_FALSE(try_byte_stream_split_decode(encoded.data(), width,
offset, count, rows,
+ decoded.data()))
+ << "width=" << width;
+ doris::byte_stream_split_decode(encoded.data(),
static_cast<int>(width), offset, count,
+ rows, decoded.data());
+ EXPECT_EQ(0, memcmp(decoded.data(), plain.data() + offset * width,
decoded.size()))
+ << "width=" << width;
+ }
+}
+
+template <typename T>
+void test_delta_prefix_sum() {
+ std::vector<T> deltas(71);
+ for (size_t row = 0; row < deltas.size(); ++row) {
+ deltas[row] = static_cast<T>((row * 7) % 19);
+ }
+ auto expected = deltas;
+ T expected_last = std::numeric_limits<T>::max() - 37;
+ constexpr T min_delta = static_cast<T>(-11);
+ using Unsigned = std::make_unsigned_t<T>;
+ for (auto& value : expected) {
+ value = static_cast<T>(static_cast<Unsigned>(value) +
static_cast<Unsigned>(min_delta) +
+ static_cast<Unsigned>(expected_last));
+ expected_last = value;
+ }
+
+ T last = std::numeric_limits<T>::max() - 37;
+ delta_decode(deltas.data(), deltas.size(), min_delta, &last);
+ EXPECT_EQ(deltas, expected);
+ EXPECT_EQ(last, expected_last);
+}
+
+TEST(ParquetSimdKernelsTest, DeltaPrefixSumPreservesParquetUnsignedOverflow) {
+ test_delta_prefix_sum<int32_t>();
+ test_delta_prefix_sum<int64_t>();
+}
+
+TEST(ParquetSimdKernelsTest, DictionaryGatherHandlesTailsAndRepeatedIds) {
+ for (const size_t width : {4, 8}) {
+ constexpr size_t entries = 257;
+ std::vector<uint8_t> dictionary(entries * width);
+ for (size_t byte = 0; byte < dictionary.size(); ++byte) {
+ dictionary[byte] = static_cast<uint8_t>((byte * 13 + 5) & 0xff);
+ }
+ std::vector<uint32_t> ids(69);
+ for (size_t row = 0; row < ids.size(); ++row) {
+ ids[row] = row % 9 == 0 ? 3 : static_cast<uint32_t>((row * 31) %
entries);
+ }
+ std::vector<uint8_t> actual(ids.size() * width);
+ dictionary_gather(dictionary.data(), ids.data(), ids.size(), width,
actual.data());
+ for (size_t row = 0; row < ids.size(); ++row) {
+ EXPECT_EQ(0, memcmp(actual.data() + row * width, dictionary.data()
+ ids[row] * width,
+ width));
+ }
+ }
+}
+
+TEST(ParquetSimdKernelsTest, NullableExpansionIsSafeForOverlappingStorage) {
+ for (const size_t width : {4, 8}) {
+ const std::vector<std::vector<uint8_t>> null_masks {
+ {1, 0, 1, 0, 0, 1, 0, 1, 0, 0, 0, 1, 1, 0, 0, 1, 0},
+ std::vector<uint8_t>(17, 0),
+ std::vector<uint8_t>(17, 1),
+ };
+ for (const auto& nulls : null_masks) {
+ const size_t compact_count =
+ static_cast<size_t>(std::count(nulls.begin(), nulls.end(),
uint8_t {0}));
+ std::vector<uint8_t> values(nulls.size() * width, 0xcd);
+ for (size_t row = 0; row < compact_count; ++row) {
+ for (size_t byte = 0; byte < width; ++byte) {
+ values[row * width + byte] =
+ static_cast<uint8_t>((row * width + byte + 1) &
0xff);
+ }
+ }
+ const auto compact = values;
+
+ expand_nullable_values(values.data(), compact_count, nulls.data(),
nulls.size(), width);
+ size_t source = 0;
+ for (size_t row = 0; row < nulls.size(); ++row) {
+ if (nulls[row] != 0) {
+ for (size_t byte = 0; byte < width; ++byte) {
+ EXPECT_EQ(values[row * width + byte], 0);
+ }
+ } else {
+ for (size_t byte = 0; byte < width; ++byte) {
+ EXPECT_EQ(values[row * width + byte], compact[source *
width + byte])
+ << "row=" << row << ", byte=" << byte << ",
width=" << width;
+ }
+ ++source;
+ }
+ }
+ }
+ }
+}
+
+template <typename T>
+void expect_raw_comparison(const std::vector<T>& values, T literal,
RawComparisonOp op,
+ const std::vector<uint8_t>& expected) {
+ std::vector<uint8_t> matches(values.size(), 1);
+ if (matches.size() > 2) {
+ matches[1] = 0;
+ }
+ auto masked_expected = expected;
+ if (masked_expected.size() > 2) {
+ masked_expected[1] = 0;
+ }
+ raw_compare(reinterpret_cast<const uint8_t*>(values.data()),
values.size(), literal, op,
+ matches.data());
+ EXPECT_EQ(matches, masked_expected);
+}
+
+TEST(ParquetSimdKernelsTest,
RawPredicatesPreserveExistingMaskAndDorisNanOrdering) {
+ const float nan = std::numeric_limits<float>::quiet_NaN();
+ const std::vector<float> floats {-3.0F, -2.0F, -1.0F, 0.0F, 1.0F, 2.0F,
+ 3.0F, nan, nan, 4.0F, 5.0F};
+ expect_raw_comparison(floats, 0.0F, RawComparisonOp::GT, {0, 0, 0, 0, 1,
1, 1, 1, 1, 1, 1});
+ expect_raw_comparison(floats, nan, RawComparisonOp::EQ, {0, 0, 0, 0, 0, 0,
0, 1, 1, 0, 0});
+ expect_raw_comparison(floats, nan, RawComparisonOp::LE, {1, 1, 1, 1, 1, 1,
1, 1, 1, 1, 1});
+
+ const std::vector<int32_t> ints {-3, -1, 0, 1, 3, 7, 9, 12, 15, 21, 27};
+ expect_raw_comparison(ints, 3, RawComparisonOp::GE, {0, 0, 0, 0, 1, 1, 1,
1, 1, 1, 1});
+ expect_raw_comparison(ints, 3, RawComparisonOp::NE, {1, 1, 1, 1, 0, 1, 1,
1, 1, 1, 1});
+
+ const double double_nan = std::numeric_limits<double>::quiet_NaN();
+ const std::vector<double> doubles {
+ -std::numeric_limits<double>::infinity(), -2.0, -1.0,
0.0, 1.0,
+ std::numeric_limits<double>::infinity(), double_nan, double_nan};
+ expect_raw_comparison(doubles, 0.0, RawComparisonOp::EQ, {0, 0, 0, 1, 0,
0, 0, 0});
+ expect_raw_comparison(doubles, 0.0, RawComparisonOp::NE, {1, 1, 1, 0, 1,
1, 1, 1});
+ expect_raw_comparison(doubles, 0.0, RawComparisonOp::LT, {1, 1, 1, 0, 0,
0, 0, 0});
+ expect_raw_comparison(doubles, 0.0, RawComparisonOp::LE, {1, 1, 1, 1, 0,
0, 0, 0});
+ expect_raw_comparison(doubles, 0.0, RawComparisonOp::GT, {0, 0, 0, 0, 1,
1, 1, 1});
+ expect_raw_comparison(doubles, 0.0, RawComparisonOp::GE, {0, 0, 0, 1, 1,
1, 1, 1});
+ expect_raw_comparison(doubles, double_nan, RawComparisonOp::EQ, {0, 0, 0,
0, 0, 0, 1, 1});
+ expect_raw_comparison(doubles, double_nan, RawComparisonOp::LT, {1, 1, 1,
1, 1, 1, 0, 0});
+ expect_raw_comparison(doubles, double_nan, RawComparisonOp::LE, {1, 1, 1,
1, 1, 1, 1, 1});
+
+ const std::vector<int64_t> bigints {-9, -3, -1, 0, 1, 3, 7, 11, 19};
+ expect_raw_comparison(bigints, int64_t {3}, RawComparisonOp::LT, {1, 1, 1,
1, 1, 0, 0, 0, 0});
+ expect_raw_comparison(bigints, int64_t {3}, RawComparisonOp::GE, {0, 0, 0,
0, 0, 1, 1, 1, 1});
+}
+
+} // namespace
+} // namespace doris::simd
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
