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 9ec7f76d15e [bench](parquet) Add native reader microbenchmark matrix
(#65921)
9ec7f76d15e is described below
commit 9ec7f76d15efc2df544b8ba6c1afdee76fb9d8c4
Author: Gabriel <[email protected]>
AuthorDate: Thu Jul 23 08:01:50 2026 +0800
[bench](parquet) Add native reader microbenchmark matrix (#65921)
## What problem does this PR solve?
Adds a reproducible local microbenchmark foundation for the native
Parquet reader described in the [benchmark
design]. It
separates page decoder cost from end-to-end local reader cost and keeps
fixture construction and encoding validation outside measured regions.
## What is changed?
- Add 152 native decoder cases covering PLAIN, dictionary,
byte-stream-split, and DELTA encodings across fixed-width and binary
physical types, with clustered and alternating sparse selections.
- Add 137 local reader cases covering open-to-first-block, full scan,
predicate scan, LIMIT shapes, null density/pattern, predicate
selectivity, projection mode, schema width, predicate position, and four
physical encodings.
- Generate deterministic fixtures lazily and validate every row group
and column encoding from Parquet footer metadata before measuring.
- Report rows/s, bytes/s, raw/selected rows, fixture size, ns/raw row,
and ns/selected row.
- Add matrix constraint tests and local build/run documentation.
## Check List
- [x] Release benchmark target builds with `ninja -C be/build_RELEASE
-j128 benchmark_test`
- [x] Scenario matrix tests pass (4/4)
- [x] All 152 decoder benchmarks pass
- [x] All 137 reader benchmarks pass
- [x] clang-format 16 and `git diff --check` pass
---
be/benchmark/benchmark_main.cpp | 39 ++
be/benchmark/parquet/AGENTS.md | 296 +++++++++++++
be/benchmark/parquet/README.md | 63 +++
be/benchmark/parquet/benchmark_parquet_decoder.hpp | 479 +++++++++++++++++++++
be/benchmark/parquet/benchmark_parquet_reader.hpp | 448 +++++++++++++++++++
be/benchmark/parquet/parquet_benchmark_scenarios.h | 250 +++++++++++
.../parquet/parquet_benchmark_scenarios_test.cpp | 129 ++++++
7 files changed, 1704 insertions(+)
diff --git a/be/benchmark/benchmark_main.cpp b/be/benchmark/benchmark_main.cpp
index d6ac9c113f4..582db25ff6d 100644
--- a/be/benchmark/benchmark_main.cpp
+++ b/be/benchmark/benchmark_main.cpp
@@ -17,6 +17,11 @@
#include <benchmark/benchmark.h>
+#include <cstdlib>
+#include <filesystem>
+#include <iostream>
+#include <vector>
+
#include "benchmark_arrow_validation.hpp"
#include "benchmark_bit_pack.hpp"
#include "benchmark_bits.hpp"
@@ -39,6 +44,8 @@
#include "core/column/column_string.h"
#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_reader.hpp"
#include "runtime/exec_env.h"
#include "runtime/memory/mem_tracker_limiter.h"
#include "runtime/memory/thread_mem_tracker_mgr.h"
@@ -52,6 +59,35 @@
namespace doris { // change if need
+static bool init_benchmark_config(const char* executable) {
+ std::vector<std::filesystem::path> candidates;
+ if (const char* doris_home = std::getenv("DORIS_HOME"); doris_home !=
nullptr) {
+ candidates.emplace_back(std::filesystem::path(doris_home) / "conf" /
"be.conf");
+ }
+ candidates.emplace_back(std::filesystem::current_path() / "conf" /
"be.conf");
+ const auto executable_dir =
std::filesystem::absolute(executable).parent_path();
+ candidates.emplace_back(executable_dir / ".." / ".." / "conf" / "be.conf");
+ candidates.emplace_back(executable_dir / ".." / ".." / ".." / "conf" /
"be.conf");
+ for (const auto& candidate : candidates) {
+ if (!std::filesystem::exists(candidate)) {
+ continue;
+ }
+ if (std::getenv("DORIS_HOME") == nullptr) {
+ const auto inferred_home =
candidate.parent_path().parent_path().string();
+ // be.conf contains paths relative to DORIS_HOME. Keep standalone
benchmark launches
+ // equivalent to build/test scripts after locating the same
repository config.
+ if (::setenv("DORIS_HOME", inferred_home.c_str(), 0) != 0) {
+ return false;
+ }
+ }
+ // Mutable config storage is populated by config::init. Reader
benchmarks must use the
+ // production defaults because zero-initialized safety limits reject
every valid footer.
+ return config::init(candidate.c_str(), false);
+ }
+ std::cerr << "Unable to find conf/be.conf for benchmark initialization\n";
+ return false;
+}
+
static void Example1(benchmark::State& state) {
// init. dont time it.
state.PauseTiming();
@@ -78,6 +114,9 @@ BENCHMARK(Example1);
// ThreadContext + mem tracker, otherwise the allocator throws E-7412. Mirrors
// the minimal subset of be/test/testutil/run_all_tests.cpp::main.
int main(int argc, char** argv) {
+ if (!doris::init_benchmark_config(argv[0])) {
+ return 1;
+ }
doris::config::enable_bmi2_optimizations = true;
SCOPED_INIT_THREAD_CONTEXT();
diff --git a/be/benchmark/parquet/AGENTS.md b/be/benchmark/parquet/AGENTS.md
new file mode 100644
index 00000000000..a8667fb43cd
--- /dev/null
+++ b/be/benchmark/parquet/AGENTS.md
@@ -0,0 +1,296 @@
+# Parquet microbenchmark guide for agents
+
+This file applies to `be/benchmark/parquet/`. Read it before changing,
running, or interpreting
+these benchmarks. The current suite is a local benchmark foundation, not the
complete Parquet
+benchmark system described in the design document.
+
+## What exists today
+
+The benchmark binary registers two groups:
+
+- `ParquetDecoder`: native page decoder benchmarks using in-memory encoded
pages.
+- `ParquetReader`: local-file benchmarks that call the format V2 Parquet
reader directly.
+
+The relevant files are:
+
+- `benchmark_parquet_decoder.hpp`: deterministic page construction and decoder
registration.
+- `benchmark_parquet_reader.hpp`: deterministic local Parquet fixtures and
reader registration.
+- `parquet_benchmark_scenarios.h`: scenario definitions and the selected
matrix.
+- `README.md`: short human-oriented build and invocation examples.
+- `be/test/format_v2/parquet/parquet_benchmark_scenarios_test.cpp`: matrix
invariants.
+
+Do not describe this suite as end-to-end SQL, `FileScannerV2`, remote I/O,
V1/V2 comparison, or a
+cross-engine benchmark. The reader benchmark starts at
`format::parquet::ParquetReader` and does
+not include FE planning, scanner scheduling, `TableReader`, client latency, or
Runtime Profile
+collection.
+
+## Build and list cases
+
+Performance results must come from a Release build:
+
+```shell
+./build.sh --benchmark -j128
+```
+
+List all Parquet cases and verify the expected registration counts:
+
+```shell
+be/output/lib/benchmark_test --benchmark_list_tests \
+ | grep -E '^Parquet(Decoder|Reader)/'
+
+be/output/lib/benchmark_test --benchmark_list_tests \
+ | grep -c '^ParquetDecoder/' # currently 152
+
+be/output/lib/benchmark_test --benchmark_list_tests \
+ | grep -c '^ParquetReader/' # currently 137
+```
+
+When running the binary directly from `be/build_RELEASE/bin`, make sure the
JVM and third-party
+libraries are discoverable. Prefer the installed
`be/output/lib/benchmark_test` produced by
+`build.sh` because the normal Doris environment already configures its
dependencies.
+
+## Run smoke verification
+
+A smoke run proves that every registered case initializes and completes. It
does not produce a
+stable performance baseline:
+
+```shell
+be/output/lib/benchmark_test \
+ --benchmark_filter='^ParquetDecoder/' \
+ --benchmark_min_time=0.001s \
+ --benchmark_out=parquet-decoder-smoke.json \
+ --benchmark_out_format=json
+
+be/output/lib/benchmark_test \
+ --benchmark_filter='^ParquetReader/' \
+ --benchmark_min_time=0.001s \
+ --benchmark_out=parquet-reader-smoke.json \
+ --benchmark_out_format=json
+```
+
+Reject a smoke run if the process is non-zero, the expected number of JSON
results is absent, or
+any result contains `error_occurred`. Also run the scenario matrix unit test
after changing the
+matrix. Never use the 1 ms smoke measurements to claim a speedup or regression.
+
+## Run a performance comparison
+
+Compare the same named cases on the same host, compiler, build flags, CPU set,
NUMA node, and power
+policy. Use the same fixture directory and cache state for both revisions. A
suitable local command
+is:
+
+```shell
+taskset -c 8 be/output/lib/benchmark_test \
+
--benchmark_filter='^ParquetReader/predicate_scan/plain/null_50/alternating/sel_10/'
\
+ --benchmark_min_time=1s \
+ --benchmark_repetitions=10 \
+ --benchmark_report_aggregates_only=true \
+ --benchmark_out=parquet-reader.json \
+ --benchmark_out_format=json
+```
+
+Use at least three untimed warmups before collecting local warm-cache results.
Run revisions in an
+ABBA order when comparing two commits. Do not compare results from different
CPUs, cache
+topologies, compiler versions, or build types.
+
+The suite does not currently control the OS page cache. Repeated reader cases
are normally
+warm-cache measurements. Do not label them cold-NVMe results. Do not clear a
shared machine's page
+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.
+
+| Encoding | Physical types |
+|---|---|
+| PLAIN | INT32, INT64, FLOAT, DOUBLE, BYTE_ARRAY, FIXED_LEN_BYTE_ARRAY |
+| Dictionary | INT32, INT64, FLOAT, DOUBLE, BYTE_ARRAY, FIXED_LEN_BYTE_ARRAY |
+| BYTE_STREAM_SPLIT | FLOAT, DOUBLE, FIXED_LEN_BYTE_ARRAY |
+| DELTA_BINARY_PACKED | INT32, INT64 |
+| DELTA_LENGTH_BYTE_ARRAY | BYTE_ARRAY |
+| DELTA_BYTE_ARRAY | BYTE_ARRAY |
+
+`ParquetReader` deliberately uses a single-variable matrix rather than a
Cartesian product. After
+deduplication it contains 137 cases covering:
+
+- operations: open-to-first-block, full scan, predicate scan, limit 1, and
limit 1000;
+- file encodings: PLAIN, dictionary, BYTE_STREAM_SPLIT, and
DELTA_BINARY_PACKED;
+- null ratios: 0%, 1%, 10%, 50%, and 90%;
+- null shapes: clustered and alternating;
+- predicate selectivities: 0%, 1%, 10%, 50%, 90%, and 100%;
+- projection shapes: predicate-only and predicate plus one lazy payload column;
+- schema widths: 4, 32, 128, and 512 columns;
+- predicate position: first or last column.
+
+Except for the axis being varied, reader cases inherit the baseline: nullable
INT32, PLAIN,
+alternating 10% nulls, 10% selectivity, 32 columns, predicate at column zero,
and predicate plus
+payload projection.
+
+## How decoder data is generated
+
+Decoder pages are constructed in memory before the timed loop. There is no
Parquet file, Python
+generator, random seed, or manifest involved.
+
+- Every page contains 65,536 logical values.
+- Integer values use `(row * 17) % 1000003`.
+- Floating-point values use `(row % 1009) * 0.25 - 100.0`.
+- Binary values are fixed at 16 bytes. Their first bytes contain `row % 1009`
and the remaining
+ bytes are deterministic filler.
+- PLAIN fixed-width values are copied in physical byte order. PLAIN BYTE_ARRAY
values use a
+ four-byte length followed by payload bytes.
+- BYTE_STREAM_SPLIT pages transpose the byte lanes of the fixed-width PLAIN
representation.
+- DELTA pages are produced with the Arrow Parquet DELTA encoders.
+- Dictionary pages contain 256 deterministic entries. IDs repeat from 0
through 255 and are
+ encoded with an eight-bit RLE/bit-packed stream.
+- Clustered selection is one continuous range. Alternating selection spreads
selected rows evenly
+ 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.
+
+## How reader Parquet files are generated
+
+Reader fixtures are generated lazily by C++ code using Arrow builders and the
Arrow Parquet writer.
+They are stored under:
+
+```text
+${TMPDIR:-/tmp}/doris_parquet_reader_benchmark/
+```
+
+Fixture contents and writer settings are:
+
+- 16,384 rows and 4 row groups of 4,096 rows each;
+- all columns are nullable INT32 and reuse the same deterministic Arrow array;
+- every non-null value is `row % 100`;
+- the predicate is `value < selectivity_percent`, so the threshold maps
directly to the intended
+ non-null selectivity;
+- alternating nulls use a 101-row period and `(row * 37) % 101`, avoiding
direct correlation with
+ the 100-value predicate period;
+- clustered nulls use contiguous null prefixes inside each 1,024-row cluster;
+- Parquet format version 2.6, DataPage V2, no compression, and statistics
disabled;
+- dictionary is explicitly enabled only for dictionary fixtures; other
fixtures disable dictionary
+ and request their named encoding;
+- after writing, every column chunk in every row group is checked through
footer metadata to ensure
+ that the writer did not silently choose another encoding.
+
+The fixture name depends only on axes that change file contents. Therefore
predicate thresholds,
+projection mode, and operation can reuse the same file. Existing fixtures are
footer-validated
+before reuse. Delete only the dedicated `doris_parquet_reader_benchmark`
temporary directory when
+fixture rules change; never remove a broad temporary or workspace directory.
+
+Fixture creation and footer verification happen before benchmark iterations.
Reader initialization
+and `close()` are excluded from steady-state full/predicate/LIMIT timing. The
+`open_to_first_block` case intentionally includes reader construction,
metadata loading, `open()`,
+and the first `get_block()` call.
+
+## Interpret the result correctly
+
+Google Benchmark JSON contains `real_time`, `cpu_time`, repetitions, and
custom counters. Use:
+
+- `cpu_time` for decoder CPU efficiency;
+- both `real_time` and `cpu_time` for reader cases, because latency and CPU
cost answer different
+ questions;
+- median across repetitions as the primary result;
+- p95 and a confidence interval for automated gates; never select the fastest
repetition.
+
+Custom counters mean:
+
+- `raw_rows`: logical source rows considered by the case;
+- `selected_rows`: rows returned after the predicate or selection;
+- `ns/raw_row`: normalized source-row cost and the primary comparison across
selectivities;
+- `ns/selected_row`: cost per surviving row; it naturally rises as selectivity
falls;
+- `selection_ranges`: decoder physical ranges; compare alternating with
clustered at the same
+ selectivity to expose per-range overhead;
+- `fixture_bytes` or `encoded_bytes`: fixture/page size, not peak memory;
+- `items_per_second`: selected rows per second, so it is not directly
comparable across different
+ selectivities;
+- `bytes_per_second`: a logical benchmark counter. In reader cases it is
estimated from projected
+ INT32 values, not measured storage-device bytes or compressed bytes.
+
+Useful comparisons are:
+
+- clustered versus alternating at equal selectivity: sparse-range/cursor
overhead;
+- null 0% versus 1/10/50/90% at equal selection: definition-level and nullable
reconstruction
+ overhead in the reader;
+- predicate-only versus predicate-projected: lazy materialization benefit;
+- width 4/32/128/512 with predicate first/last: schema and metadata traversal
overhead;
+- PLAIN versus dictionary/BYTE_STREAM_SPLIT/DELTA at the same reader baseline:
encoding path cost;
+- open-to-first-block and LIMIT cases separately from steady-state full scans.
+
+Before attributing a regression, confirm that output row counts are identical,
the benchmark name
+and fixture encoding match, and repetition variability is acceptable. A
coefficient of variation
+above 3% is inconclusive and should be rerun. Correlate CPU regressions with
`perf stat` counters
+such as cycles, instructions, branch misses, and cache misses when possible.
+
+## Coverage audit and required follow-up
+
+The current matrix is not comprehensive. Preserve this distinction in PR
descriptions and reports.
+
+### P0: complete the local benchmark phase
+
+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.
+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.
+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.
+6. Add dictionary footprint sweeps based on actual bytes relative to L1/L2
cache, value-width
+ sweeps, cardinality sweeps, and dictionary-to-PLAIN fallback/mixed-page
files. The current
+ dictionary is small and fixed.
+7. Add a Doris V1/V2 SQL runner over the same verified corpus, correctness
comparison, Runtime
+ Profile collection, normalized result JSON, comparison report, and the
first stable local
+ baseline. The current 1 ms run only proves executability.
+
+### P1: broaden Parquet format and execution behavior
+
+1. Vary compression codecs: uncompressed, Snappy, ZSTD, LZ4, Brotli, and Gzip
where supported.
+2. Cover DataPage V1 and V2, page size, row-group size/count, page boundaries,
small files, and
+ large files that exceed cache capacity.
+3. Add required/non-nullable columns, correlated and anti-correlated
null/predicate distributions,
+ random deterministic runs, long null runs, and selections that cross page
boundaries.
+4. Add Boolean, INT96 compatibility, logical annotations, timestamp units/time
zones, decimal
+ physical representations, short/long strings, skew, and empty values.
+5. Add nested ARRAY/MAP/STRUCT and repeated definition/repetition-level cases.
+6. Add row-group statistics, page index, Bloom filter, dictionary filtering,
all-filtered batches,
+ missing-column/default-value handling, schema mapping, and schema-evolution
cases. Keep pruning
+ benchmarks separate from decode benchmarks so skipped work is not
misattributed to a faster
+ decoder.
+7. Add predicate LIMIT 1/1000 cases. Current LIMIT cases only vary reader
batch size without a
+ predicate.
+8. Add peak tracked memory, allocation counts, and an explicit warm/cold-cache
methodology.
+
+### Later phases
+
+Remote S3/FileCache/RTT workloads, prefetch metrics, cross-engine adapters,
nightly/weekly jobs, and
+release regression gates belong to later phases. They require isolated
infrastructure and must not
+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.
+
+## Rules for extending the suite
+
+- Keep deterministic data construction and validation outside timed regions.
+- Verify actual footer encodings; never trust writer configuration alone.
+- Change one primary axis at a time and add only meaningful second-order
interactions.
+- Add or update matrix invariant tests before registering new dimensions.
+- Keep correctness/error-injection tests separate from throughput benchmarks.
+- Do not commit large generated corpora. Commit generation rules and
manifests; keep only small
+ correctness fixtures in the repository when necessary.
+- Record the exact commit, compiler, build type, host topology, command,
repetitions, cache state,
+ and raw JSON for any claimed performance result.
+- Never claim a performance improvement from a smoke run or from incomparable
machines.
diff --git a/be/benchmark/parquet/README.md b/be/benchmark/parquet/README.md
new file mode 100644
index 00000000000..40c79bc28be
--- /dev/null
+++ b/be/benchmark/parquet/README.md
@@ -0,0 +1,63 @@
+# Parquet reader microbenchmarks
+
+These benchmarks separate native page decoding from the complete local-file
reader path. They use
+deterministic data and verify the physical encoding recorded in each generated
Parquet footer before
+running a measurement.
+
+Agents and maintainers must read [AGENTS.md](AGENTS.md) for the exact timing
boundaries, synthetic
+data rules, result interpretation, current coverage limitations, and
prioritized follow-up work.
+
+## Build
+
+Build the Release benchmark binary from the repository root:
+
+```shell
+./build.sh --benchmark -j128
+```
+
+List only the Parquet cases:
+
+```shell
+be/output/lib/benchmark_test --benchmark_list_tests | grep '^Parquet'
+```
+
+## Decoder cases
+
+`ParquetDecoder` measures the native decoder with data generation and encoder
setup outside the
+timed region. It covers PLAIN, dictionary, byte-stream-split, and DELTA
encodings across their
+supported fixed-width and binary physical types. Sparse selections are
provided as both one
+clustered range and many alternating ranges.
+
+```shell
+be/output/lib/benchmark_test \
+ --benchmark_filter='^ParquetDecoder/plain/int64/sel_10/alternating$' \
+ --benchmark_min_time=0.1s
+```
+
+## Local reader cases
+
+`ParquetReader` measures local open-to-first-block, full scan, predicate scan,
and LIMIT-shaped
+reads. The matrix covers:
+
+- PLAIN, dictionary, byte-stream-split, and DELTA binary-packed files;
+- NULL ratios of 0%, 1%, 10%, 50%, and 90%, with clustered and alternating
placement;
+- predicate selectivities of 0%, 1%, 10%, 50%, 90%, and 100%;
+- predicate-only and predicate-plus-lazy-projected reads;
+- schemas with 4, 32, 128, and 512 columns, with the predicate first or last.
+
+Fixtures are created lazily under the system temporary directory in
+`doris_parquet_reader_benchmark`. Generation, footer validation, and reader
setup are excluded from
+steady-state scan timings. `open_to_first_block` intentionally includes reader
initialization,
+footer loading, open, and the first `get_block` call.
+
+```shell
+be/output/lib/benchmark_test \
+
--benchmark_filter='^ParquetReader/predicate_scan/plain/null_50/alternating/sel_10/'
\
+ --benchmark_min_time=0.1s \
+ --benchmark_out=parquet-reader.json \
+ --benchmark_out_format=json
+```
+
+Every result reports throughput plus `raw_rows`, `selected_rows`,
`fixture_bytes`, `ns/raw_row`,
+and (when at least one row survives) `ns/selected_row`. Keep CPU frequency,
build type, compiler,
+machine placement, and benchmark filters fixed when comparing two commits.
diff --git a/be/benchmark/parquet/benchmark_parquet_decoder.hpp
b/be/benchmark/parquet/benchmark_parquet_decoder.hpp
new file mode 100644
index 00000000000..25fb05d1449
--- /dev/null
+++ b/be/benchmark/parquet/benchmark_parquet_decoder.hpp
@@ -0,0 +1,479 @@
+// 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 <parquet/encoding.h>
+#include <parquet/schema.h>
+
+#include <algorithm>
+#include <array>
+#include <cstdint>
+#include <cstring>
+#include <memory>
+#include <numeric>
+#include <stdexcept>
+#include <string>
+#include <type_traits>
+#include <vector>
+
+#include "core/custom_allocator.h"
+#include "format_v2/parquet/reader/native/decoder.h"
+#include "parquet_benchmark_scenarios.h"
+#include "util/faststring.h"
+#include "util/rle_encoding.h"
+#include "util/slice.h"
+
+namespace doris::parquet_benchmark {
+namespace detail {
+
+constexpr size_t DECODER_ROWS = 1UL << 16;
+constexpr size_t DICTIONARY_ENTRIES = 256;
+constexpr size_t FIXED_BINARY_WIDTH = 16;
+
+struct EncodedPage {
+ std::vector<uint8_t> data;
+ std::vector<uint8_t> dictionary;
+ size_t dictionary_entries = 0;
+ size_t value_width = 0;
+ bool binary = false;
+};
+
+inline tparquet::Type::type physical_type(ValueType value_type) {
+ switch (value_type) {
+ case ValueType::INT32:
+ return tparquet::Type::INT32;
+ case ValueType::INT64:
+ return tparquet::Type::INT64;
+ case ValueType::FLOAT:
+ return tparquet::Type::FLOAT;
+ case ValueType::DOUBLE:
+ return tparquet::Type::DOUBLE;
+ case ValueType::BYTE_ARRAY:
+ return tparquet::Type::BYTE_ARRAY;
+ case ValueType::FIXED_LEN_BYTE_ARRAY:
+ return tparquet::Type::FIXED_LEN_BYTE_ARRAY;
+ }
+ throw std::logic_error("unknown Parquet benchmark value type");
+}
+
+inline tparquet::Encoding::type parquet_encoding(Encoding encoding) {
+ switch (encoding) {
+ case Encoding::PLAIN:
+ return tparquet::Encoding::PLAIN;
+ case Encoding::DICTIONARY:
+ return tparquet::Encoding::RLE_DICTIONARY;
+ case Encoding::BYTE_STREAM_SPLIT:
+ return tparquet::Encoding::BYTE_STREAM_SPLIT;
+ case Encoding::DELTA_BINARY_PACKED:
+ return tparquet::Encoding::DELTA_BINARY_PACKED;
+ case Encoding::DELTA_LENGTH_BYTE_ARRAY:
+ return tparquet::Encoding::DELTA_LENGTH_BYTE_ARRAY;
+ case Encoding::DELTA_BYTE_ARRAY:
+ return tparquet::Encoding::DELTA_BYTE_ARRAY;
+ }
+ throw std::logic_error("unknown Parquet benchmark encoding");
+}
+
+inline std::shared_ptr<::parquet::ColumnDescriptor>
descriptor(::parquet::Type::type type,
+ int type_length
= -1) {
+ auto node = type == ::parquet::Type::FIXED_LEN_BYTE_ARRAY
+ ? ::parquet::schema::PrimitiveNode::Make(
+ "value", ::parquet::Repetition::REQUIRED,
type,
+ ::parquet::ConvertedType::NONE, type_length)
+ : ::parquet::schema::PrimitiveNode::Make(
+ "value", ::parquet::Repetition::REQUIRED,
type);
+ return std::make_shared<::parquet::ColumnDescriptor>(node, 0, 0);
+}
+
+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);
+ }
+ }
+ return values;
+}
+
+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));
+ }
+ return values;
+}
+
+inline std::vector<uint8_t> encode_plain_binary(const
std::vector<std::string>& values) {
+ size_t bytes = 0;
+ for (const auto& value : values) {
+ bytes += sizeof(uint32_t) + value.size();
+ }
+ std::vector<uint8_t> encoded;
+ encoded.reserve(bytes);
+ for (const auto& value : values) {
+ const auto length = static_cast<uint32_t>(value.size());
+ const auto* length_bytes = reinterpret_cast<const uint8_t*>(&length);
+ encoded.insert(encoded.end(), length_bytes, length_bytes +
sizeof(length));
+ encoded.insert(encoded.end(), value.begin(), value.end());
+ }
+ return encoded;
+}
+
+template <typename T>
+std::vector<uint8_t> encode_plain_fixed(const std::vector<T>& values) {
+ std::vector<uint8_t> encoded(values.size() * sizeof(T));
+ memcpy(encoded.data(), values.data(), encoded.size());
+ return encoded;
+}
+
+inline std::vector<uint8_t> encode_fixed_binary(const
std::vector<std::string>& values) {
+ std::vector<uint8_t> encoded;
+ encoded.reserve(values.size() * FIXED_BINARY_WIDTH);
+ for (const auto& value : values) {
+ encoded.insert(encoded.end(), value.begin(), value.end());
+ }
+ return encoded;
+}
+
+inline std::vector<uint8_t> encode_byte_stream_split(const
std::vector<uint8_t>& plain,
+ size_t value_width) {
+ const size_t rows = plain.size() / value_width;
+ std::vector<uint8_t> encoded(plain.size());
+ for (size_t row = 0; row < rows; ++row) {
+ for (size_t byte = 0; byte < value_width; ++byte) {
+ encoded[byte * rows + row] = plain[row * value_width + byte];
+ }
+ }
+ return encoded;
+}
+
+template <typename ParquetType, typename T>
+std::vector<uint8_t> encode_delta_fixed(const std::vector<T>& values) {
+ auto desc = descriptor(ParquetType::type_num);
+ auto encoder = ::parquet::MakeTypedEncoder<ParquetType>(
+ ::parquet::Encoding::DELTA_BINARY_PACKED, false, desc.get());
+ encoder->Put(values.data(), static_cast<int>(values.size()));
+ const auto buffer = encoder->FlushValues();
+ return {buffer->data(), buffer->data() + buffer->size()};
+}
+
+inline std::vector<uint8_t> encode_delta_binary(const
std::vector<std::string>& values,
+ ::parquet::Encoding::type
encoding) {
+ std::vector<::parquet::ByteArray> arrays;
+ arrays.reserve(values.size());
+ for (const auto& value : values) {
+ arrays.emplace_back(static_cast<uint32_t>(value.size()),
+ reinterpret_cast<const uint8_t*>(value.data()));
+ }
+ auto desc = descriptor(::parquet::Type::BYTE_ARRAY);
+ auto encoder =
+ ::parquet::MakeTypedEncoder<::parquet::ByteArrayType>(encoding,
false, desc.get());
+ encoder->Put(arrays.data(), static_cast<int>(arrays.size()));
+ const auto buffer = encoder->FlushValues();
+ return {buffer->data(), buffer->data() + buffer->size()};
+}
+
+inline std::vector<uint8_t> encode_dictionary_indices(size_t rows) {
+ faststring encoded_ids;
+ RleEncoder<uint32_t> encoder(&encoded_ids, 8);
+ for (size_t row = 0; row < rows; ++row) {
+ encoder.Put(static_cast<uint32_t>(row % DICTIONARY_ENTRIES));
+ }
+ for (size_t padding = rows; padding % 8 != 0; ++padding) {
+ encoder.Put(0);
+ }
+ encoder.Flush();
+ std::vector<uint8_t> result(encoded_ids.size() + 1);
+ result[0] = 8;
+ memcpy(result.data() + 1, encoded_ids.data(), encoded_ids.size());
+ return result;
+}
+
+inline EncodedPage dictionary_page(ValueType value_type) {
+ EncodedPage page;
+ page.data = encode_dictionary_indices(DECODER_ROWS);
+ page.dictionary_entries = DICTIONARY_ENTRIES;
+ page.binary = value_type == ValueType::BYTE_ARRAY;
+ switch (value_type) {
+ case ValueType::INT32:
+ page.value_width = sizeof(int32_t);
+ page.dictionary =
encode_plain_fixed(fixed_values<int32_t>(DICTIONARY_ENTRIES));
+ break;
+ case ValueType::INT64:
+ page.value_width = sizeof(int64_t);
+ page.dictionary =
encode_plain_fixed(fixed_values<int64_t>(DICTIONARY_ENTRIES));
+ break;
+ case ValueType::FLOAT:
+ page.value_width = sizeof(float);
+ page.dictionary =
encode_plain_fixed(fixed_values<float>(DICTIONARY_ENTRIES));
+ break;
+ case ValueType::DOUBLE:
+ page.value_width = sizeof(double);
+ page.dictionary =
encode_plain_fixed(fixed_values<double>(DICTIONARY_ENTRIES));
+ break;
+ case ValueType::BYTE_ARRAY:
+ page.value_width = FIXED_BINARY_WIDTH;
+ page.dictionary =
encode_plain_binary(binary_values(DICTIONARY_ENTRIES));
+ break;
+ case ValueType::FIXED_LEN_BYTE_ARRAY:
+ page.value_width = FIXED_BINARY_WIDTH;
+ page.dictionary =
encode_fixed_binary(binary_values(DICTIONARY_ENTRIES));
+ break;
+ }
+ return page;
+}
+
+inline EncodedPage encoded_page(const DecoderScenario& scenario) {
+ if (scenario.encoding == Encoding::DICTIONARY) {
+ return dictionary_page(scenario.value_type);
+ }
+
+ EncodedPage page;
+ page.binary = scenario.value_type == ValueType::BYTE_ARRAY;
+ switch (scenario.value_type) {
+ case ValueType::INT32: {
+ auto values = fixed_values<int32_t>(DECODER_ROWS);
+ page.value_width = sizeof(int32_t);
+ page.data = scenario.encoding == Encoding::DELTA_BINARY_PACKED
+ ? encode_delta_fixed<::parquet::Int32Type>(values)
+ : encode_plain_fixed(values);
+ break;
+ }
+ case ValueType::INT64: {
+ auto values = fixed_values<int64_t>(DECODER_ROWS);
+ page.value_width = sizeof(int64_t);
+ page.data = scenario.encoding == Encoding::DELTA_BINARY_PACKED
+ ? encode_delta_fixed<::parquet::Int64Type>(values)
+ : encode_plain_fixed(values);
+ break;
+ }
+ case ValueType::FLOAT: {
+ auto plain = encode_plain_fixed(fixed_values<float>(DECODER_ROWS));
+ page.value_width = sizeof(float);
+ page.data = scenario.encoding == Encoding::BYTE_STREAM_SPLIT
+ ? encode_byte_stream_split(plain, page.value_width)
+ : std::move(plain);
+ break;
+ }
+ case ValueType::DOUBLE: {
+ auto plain = encode_plain_fixed(fixed_values<double>(DECODER_ROWS));
+ page.value_width = sizeof(double);
+ page.data = scenario.encoding == Encoding::BYTE_STREAM_SPLIT
+ ? encode_byte_stream_split(plain, page.value_width)
+ : std::move(plain);
+ break;
+ }
+ case ValueType::BYTE_ARRAY: {
+ auto values = binary_values(DECODER_ROWS);
+ page.value_width = FIXED_BINARY_WIDTH;
+ if (scenario.encoding == Encoding::DELTA_LENGTH_BYTE_ARRAY) {
+ page.data = encode_delta_binary(values,
::parquet::Encoding::DELTA_LENGTH_BYTE_ARRAY);
+ } else if (scenario.encoding == Encoding::DELTA_BYTE_ARRAY) {
+ page.data = encode_delta_binary(values,
::parquet::Encoding::DELTA_BYTE_ARRAY);
+ } else {
+ page.data = encode_plain_binary(values);
+ }
+ break;
+ }
+ case ValueType::FIXED_LEN_BYTE_ARRAY: {
+ auto plain = encode_fixed_binary(binary_values(DECODER_ROWS));
+ page.value_width = FIXED_BINARY_WIDTH;
+ page.data = scenario.encoding == Encoding::BYTE_STREAM_SPLIT
+ ? encode_byte_stream_split(plain, page.value_width)
+ : std::move(plain);
+ break;
+ }
+ }
+ return page;
+}
+
+class FixedSink final : public ParquetFixedValueConsumer {
+public:
+ Status consume(const uint8_t* values, size_t num_values, size_t
value_width) override {
+ benchmark::DoNotOptimize(values);
+ benchmark::DoNotOptimize(num_values);
+ benchmark::DoNotOptimize(value_width);
+ consumed += num_values;
+ return Status::OK();
+ }
+
+ size_t consumed = 0;
+};
+
+class BinarySink final : public ParquetBinaryValueConsumer {
+public:
+ Status consume(const StringRef* values, size_t num_values) override {
+ benchmark::DoNotOptimize(values);
+ benchmark::DoNotOptimize(num_values);
+ consumed += num_values;
+ 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 {
+ benchmark::DoNotOptimize(encoded_data);
+ benchmark::DoNotOptimize(payload_offsets);
+ benchmark::DoNotOptimize(value_offsets);
+ auto value_spans_data = value_spans.data();
+ benchmark::DoNotOptimize(value_spans_data);
+ consumed += num_values;
+ return Status::OK();
+ }
+
+ size_t consumed = 0;
+};
+
+class DictionarySink final : public ParquetDictionaryValueConsumer {
+public:
+ DictionarySink(const uint8_t* dictionary, size_t value_width)
+ : _dictionary(dictionary), _value_width(value_width) {}
+
+ Status consume_indices(const uint32_t* indices, size_t num_values)
override {
+ for (size_t row = 0; row < num_values; ++row) {
+ auto* value = _dictionary + indices[row] * _value_width;
+ benchmark::DoNotOptimize(value);
+ }
+ consumed += num_values;
+ return Status::OK();
+ }
+
+ Status consume_repeated(uint32_t index, size_t num_values) override {
+ auto* value = _dictionary + index * _value_width;
+ benchmark::DoNotOptimize(value);
+ consumed += num_values;
+ return Status::OK();
+ }
+
+ size_t consumed = 0;
+
+private:
+ const uint8_t* _dictionary;
+ const size_t _value_width;
+};
+
+inline ParquetSelection native_selection(const SelectionPlan& plan) {
+ ParquetSelection selection {
+ .total_values = plan.total_rows, .selected_values =
plan.selected_rows, .ranges = {}};
+ selection.ranges.reserve(plan.ranges.size());
+ for (const auto& range : plan.ranges) {
+ selection.ranges.push_back({.first = range.first, .count =
range.count});
+ }
+ return selection;
+}
+
+inline void run_decoder(benchmark::State& state, DecoderScenario scenario, int
selectivity,
+ Pattern pattern) {
+ auto page = encoded_page(scenario);
+ const auto plan = make_selection_plan(DECODER_ROWS, selectivity, pattern);
+ const auto selection = native_selection(plan);
+ std::unique_ptr<format::parquet::native::Decoder> decoder;
+ auto status = format::parquet::native::Decoder::get_decoder(
+ physical_type(scenario.value_type),
parquet_encoding(scenario.encoding), decoder);
+ if (!status.ok()) {
+ state.SkipWithError(status.to_string().c_str());
+ return;
+ }
+ decoder->set_type_length(static_cast<int32_t>(page.value_width));
+ decoder->set_expected_values(DECODER_ROWS);
+
+ std::vector<uint8_t> dictionary_for_sink;
+ if (scenario.encoding == Encoding::DICTIONARY) {
+ dictionary_for_sink = page.dictionary;
+ auto dictionary = make_unique_buffer<uint8_t>(page.dictionary.size());
+ memcpy(dictionary.get(), page.dictionary.data(),
page.dictionary.size());
+ status = decoder->set_dict(dictionary,
static_cast<int32_t>(page.dictionary.size()),
+ page.dictionary_entries);
+ if (!status.ok()) {
+ state.SkipWithError(status.to_string().c_str());
+ return;
+ }
+ }
+
+ Slice encoded(page.data.data(), page.data.size());
+ FixedSink fixed_sink;
+ BinarySink binary_sink;
+ DictionarySink dictionary_sink(dictionary_for_sink.data(),
page.value_width);
+ for (auto _ : state) {
+ state.PauseTiming();
+ status = decoder->set_data(&encoded);
+ fixed_sink.consumed = 0;
+ binary_sink.consumed = 0;
+ dictionary_sink.consumed = 0;
+ state.ResumeTiming();
+ if (scenario.encoding == Encoding::DICTIONARY) {
+ status = decoder->decode_selected_dictionary_values(selection,
dictionary_sink);
+ } else if (page.binary) {
+ status = decoder->decode_selected_binary_values(selection,
binary_sink);
+ } else {
+ status = decoder->decode_selected_fixed_values(selection,
fixed_sink);
+ }
+ if (!status.ok()) {
+ state.SkipWithError(status.to_string().c_str());
+ break;
+ }
+ benchmark::ClobberMemory();
+ }
+
+ state.SetItemsProcessed(static_cast<int64_t>(state.iterations()) *
+ static_cast<int64_t>(plan.selected_rows));
+ state.SetBytesProcessed(static_cast<int64_t>(state.iterations()) *
+ static_cast<int64_t>(page.data.size()));
+ state.counters["raw_rows"] = static_cast<double>(DECODER_ROWS);
+ state.counters["selected_rows"] = static_cast<double>(plan.selected_rows);
+ state.counters["selection_ranges"] =
static_cast<double>(plan.ranges.size());
+ state.counters["encoded_bytes"] = static_cast<double>(page.data.size());
+ state.counters["ns/raw_row"] = benchmark::Counter(
+ static_cast<double>(DECODER_ROWS),
+ benchmark::Counter::kIsIterationInvariantRate |
benchmark::Counter::kInvert);
+ if (plan.selected_rows > 0) {
+ state.counters["ns/selected_row"] = benchmark::Counter(
+ static_cast<double>(plan.selected_rows),
+ benchmark::Counter::kIsIterationInvariantRate |
benchmark::Counter::kInvert);
+ }
+}
+
+inline bool register_decoder_benchmarks() {
+ for (const auto& scenario : decoder_scenarios()) {
+ for (const int selectivity : {1, 10, 50, 100}) {
+ for (const auto pattern : {Pattern::CLUSTERED,
Pattern::ALTERNATING}) {
+ const std::string name = "ParquetDecoder/" +
to_string(scenario.encoding) + "/" +
+ to_string(scenario.value_type) +
"/sel_" +
+ std::to_string(selectivity) + "/" +
to_string(pattern);
+ benchmark::RegisterBenchmark(name.c_str(),
[=](benchmark::State& state) {
+ run_decoder(state, scenario, selectivity, pattern);
+ })->Unit(benchmark::kNanosecond);
+ }
+ }
+ }
+ return true;
+}
+
+inline const bool DECODER_BENCHMARKS_REGISTERED =
register_decoder_benchmarks();
+
+} // namespace detail
+} // namespace doris::parquet_benchmark
diff --git a/be/benchmark/parquet/benchmark_parquet_reader.hpp
b/be/benchmark/parquet/benchmark_parquet_reader.hpp
new file mode 100644
index 00000000000..ae3e14c0f66
--- /dev/null
+++ b/be/benchmark/parquet/benchmark_parquet_reader.hpp
@@ -0,0 +1,448 @@
+// 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 <arrow/api.h>
+#include <arrow/io/api.h>
+#include <benchmark/benchmark.h>
+#include <parquet/api/reader.h>
+#include <parquet/arrow/writer.h>
+
+#include <algorithm>
+#include <cstdint>
+#include <filesystem>
+#include <memory>
+#include <mutex>
+#include <set>
+#include <stdexcept>
+#include <string>
+#include <vector>
+
+#include "core/assert_cast.h"
+#include "core/block/block.h"
+#include "core/column/column_nullable.h"
+#include "core/column/column_vector.h"
+#include "core/data_type/data_type_nullable.h"
+#include "core/data_type/data_type_number.h"
+#include "exprs/vexpr.h"
+#include "exprs/vexpr_context.h"
+#include "format_v2/file_reader.h"
+#include "format_v2/parquet/parquet_reader.h"
+#include "gen_cpp/Types_types.h"
+#include "io/io_common.h"
+#include "parquet_benchmark_scenarios.h"
+#include "runtime/runtime_state.h"
+#include "storage/index/zone_map/zonemap_eval_context.h"
+#include "storage/index/zone_map/zonemap_filter_result.h"
+
+namespace doris::parquet_benchmark {
+namespace reader_detail {
+
+constexpr size_t READER_ROWS = 1UL << 14;
+constexpr size_t READER_ROW_GROUP_ROWS = 1UL << 12;
+
+inline void throw_if_error(const Status& status) {
+ if (!status.ok()) {
+ throw std::runtime_error(status.to_string());
+ }
+}
+
+inline bool is_null_row(size_t row, int null_percent, Pattern pattern) {
+ if (null_percent <= 0) {
+ return false;
+ }
+ if (pattern == Pattern::ALTERNATING) {
+ constexpr size_t NULL_PERIOD = 101;
+ const size_t null_slots = (static_cast<size_t>(null_percent) *
NULL_PERIOD + 99) / 100;
+ return (row * 37) % NULL_PERIOD < null_slots;
+ }
+ constexpr size_t CLUSTER_ROWS = 1024;
+ return row % CLUSTER_ROWS < CLUSTER_ROWS *
static_cast<size_t>(null_percent) / 100;
+}
+
+inline std::shared_ptr<arrow::Array> build_int32_array(int null_percent,
Pattern pattern) {
+ arrow::Int32Builder builder;
+ PARQUET_THROW_NOT_OK(builder.Reserve(READER_ROWS));
+ for (size_t row = 0; row < READER_ROWS; ++row) {
+ if (is_null_row(row, null_percent, pattern)) {
+ PARQUET_THROW_NOT_OK(builder.AppendNull());
+ } else {
+ PARQUET_THROW_NOT_OK(builder.Append(static_cast<int32_t>(row %
100)));
+ }
+ }
+ return builder.Finish().ValueOrDie();
+}
+
+inline ::parquet::Encoding::type file_encoding(Encoding encoding) {
+ switch (encoding) {
+ case Encoding::PLAIN:
+ return ::parquet::Encoding::PLAIN;
+ case Encoding::BYTE_STREAM_SPLIT:
+ return ::parquet::Encoding::BYTE_STREAM_SPLIT;
+ case Encoding::DELTA_BINARY_PACKED:
+ return ::parquet::Encoding::DELTA_BINARY_PACKED;
+ case Encoding::DICTIONARY:
+ return ::parquet::Encoding::RLE_DICTIONARY;
+ case Encoding::DELTA_LENGTH_BYTE_ARRAY:
+ return ::parquet::Encoding::DELTA_LENGTH_BYTE_ARRAY;
+ case Encoding::DELTA_BYTE_ARRAY:
+ return ::parquet::Encoding::DELTA_BYTE_ARRAY;
+ }
+ throw std::logic_error("unknown Parquet benchmark encoding");
+}
+
+inline std::string fixture_name(const ReaderScenario& scenario) {
+ return "v2_" + to_string(scenario.encoding) + "_null" +
std::to_string(scenario.null_percent) +
+ "_" + to_string(scenario.null_pattern) + "_w" +
std::to_string(scenario.schema_width) +
+ "_p" + std::to_string(scenario.predicate_position) + ".parquet";
+}
+
+inline void verify_fixture_encoding(const std::filesystem::path& path,
+ const ReaderScenario& scenario) {
+ auto reader = ::parquet::ParquetFileReader::OpenFile(path.string(), false);
+ const auto metadata = reader->metadata();
+ if (metadata->num_rows() != static_cast<int64_t>(READER_ROWS) ||
+ metadata->num_columns() != scenario.schema_width) {
+ throw std::runtime_error("Parquet benchmark fixture has unexpected
shape: " +
+ path.string());
+ }
+ const auto expected = file_encoding(scenario.encoding);
+ for (int row_group = 0; row_group < metadata->num_row_groups();
++row_group) {
+ for (int column = 0; column < metadata->num_columns(); ++column) {
+ const auto encodings =
metadata->RowGroup(row_group)->ColumnChunk(column)->encodings();
+ if (std::ranges::find(encodings, expected) == encodings.end()) {
+ throw std::runtime_error("Parquet benchmark fixture did not
use " +
+ to_string(scenario.encoding) +
+ " encoding: " + path.string());
+ }
+ }
+ }
+}
+
+inline std::filesystem::path ensure_fixture(const ReaderScenario& scenario) {
+ static std::mutex fixture_mutex;
+ const auto directory =
+ std::filesystem::temp_directory_path() /
"doris_parquet_reader_benchmark";
+ const auto path = directory / fixture_name(scenario);
+ std::lock_guard guard(fixture_mutex);
+ if (std::filesystem::exists(path)) {
+ verify_fixture_encoding(path, scenario);
+ return path;
+ }
+
+ std::filesystem::create_directories(directory);
+ const auto temporary_path = path.string() + ".tmp";
+ std::filesystem::remove(temporary_path);
+ const auto values = build_int32_array(scenario.null_percent,
scenario.null_pattern);
+ std::vector<std::shared_ptr<arrow::Field>> fields;
+ std::vector<std::shared_ptr<arrow::ChunkedArray>> columns;
+ fields.reserve(scenario.schema_width);
+ columns.reserve(scenario.schema_width);
+ for (int column = 0; column < scenario.schema_width; ++column) {
+ fields.push_back(arrow::field("c" + std::to_string(column),
arrow::int32(), true));
+ columns.push_back(std::make_shared<arrow::ChunkedArray>(values));
+ }
+ const auto table = arrow::Table::Make(arrow::schema(std::move(fields)),
std::move(columns));
+
+ const auto output_result =
arrow::io::FileOutputStream::Open(temporary_path);
+ if (!output_result.ok()) {
+ throw std::runtime_error(output_result.status().ToString());
+ }
+ const auto output = *output_result;
+ ::parquet::WriterProperties::Builder properties;
+ properties.version(::parquet::ParquetVersion::PARQUET_2_6);
+ properties.data_page_version(::parquet::ParquetDataPageVersion::V2);
+ properties.compression(::parquet::Compression::UNCOMPRESSED);
+ properties.disable_statistics();
+ if (scenario.encoding == Encoding::DICTIONARY) {
+ properties.enable_dictionary();
+ } else {
+ properties.disable_dictionary();
+ properties.encoding(file_encoding(scenario.encoding));
+ }
+ PARQUET_THROW_NOT_OK(::parquet::arrow::WriteTable(*table,
arrow::default_memory_pool(), output,
+ READER_ROW_GROUP_ROWS,
properties.build()));
+ PARQUET_THROW_NOT_OK(output->Close());
+ std::filesystem::rename(temporary_path, path);
+ verify_fixture_encoding(path, scenario);
+ return path;
+}
+
+class Int32LessThanExpr final : public VExpr {
+public:
+ Int32LessThanExpr(int column_id, int32_t upper_bound)
+ : VExpr(std::make_shared<DataTypeUInt8>(), false),
+ _column_id(column_id),
+ _upper_bound(upper_bound) {}
+
+ Status execute_column_impl(VExprContext*, const Block* block, const
Selector* selector,
+ size_t count, ColumnPtr& result_column) const
override {
+ DORIS_CHECK(block != nullptr);
+ const auto& nullable =
+ assert_cast<const
ColumnNullable&>(*block->get_by_position(_column_id).column);
+ const auto& values = assert_cast<const
ColumnInt32&>(nullable.get_nested_column());
+ const auto& nulls = nullable.get_null_map_data();
+ auto result = ColumnUInt8::create(count, 0);
+ auto& matches = result->get_data();
+ for (size_t row = 0; row < count; ++row) {
+ const size_t input_row = selector == nullptr ? row :
(*selector)[row];
+ matches[row] = !nulls[input_row] && values.get_element(input_row)
< _upper_bound;
+ }
+ result_column = std::move(result);
+ return Status::OK();
+ }
+
+ bool can_execute_on_raw_fixed_values(const DataTypePtr& data_type,
+ int column_id) const override {
+ return column_id == _column_id &&
+ remove_nullable(data_type)->get_primitive_type() == TYPE_INT;
+ }
+
+ Status execute_on_raw_fixed_values(const uint8_t* values, size_t
num_values, size_t value_width,
+ const DataTypePtr&, int column_id,
+ uint8_t* matches) const override {
+ DORIS_CHECK(column_id == _column_id);
+ DORIS_CHECK(value_width == sizeof(int32_t));
+ const auto* typed_values = reinterpret_cast<const int32_t*>(values);
+ for (size_t row = 0; row < num_values; ++row) {
+ matches[row] &= typed_values[row] < _upper_bound;
+ }
+ return Status::OK();
+ }
+
+ bool can_evaluate_zonemap_filter() const override { return true; }
+
+ ZoneMapFilterResult evaluate_zonemap_filter(const ZoneMapEvalContext& ctx)
const override {
+ const auto zone_map = ctx.zone_map(_column_id);
+ if (zone_map == nullptr) {
+ return unsupported_zonemap_filter(ctx);
+ }
+ if (!zone_map->has_not_null) {
+ return ZoneMapFilterResult::kNoMatch;
+ }
+ const auto upper_bound = Field::create_field<TYPE_INT>(_upper_bound);
+ return zone_map->min_value >= upper_bound ?
ZoneMapFilterResult::kNoMatch
+ :
ZoneMapFilterResult::kMayMatch;
+ }
+
+ bool can_evaluate_dictionary_filter() const override { return true; }
+
+ ZoneMapFilterResult evaluate_dictionary_filter(
+ const DictionaryEvalContext& ctx) const override {
+ const auto* dictionary = ctx.slot(_column_id);
+ if (dictionary == nullptr) {
+ return ZoneMapFilterResult::kUnsupported;
+ }
+ const auto upper_bound = Field::create_field<TYPE_INT>(_upper_bound);
+ return std::ranges::any_of(dictionary->values,
+ [&](const Field& value) { return value <
upper_bound; })
+ ? ZoneMapFilterResult::kMayMatch
+ : ZoneMapFilterResult::kNoMatch;
+ }
+
+ void collect_slot_column_ids(std::set<int>& column_ids) const override {
+ column_ids.insert(_column_id);
+ }
+
+ const std::string& expr_name() const override { return _expr_name; }
+
+private:
+ const int _column_id;
+ const int32_t _upper_bound;
+ const std::string _expr_name = "ParquetBenchmarkInt32LessThan";
+};
+
+inline VExprContextSPtr make_predicate(int column_position, int
selectivity_percent) {
+ auto context = VExprContext::create_shared(
+ std::make_shared<Int32LessThanExpr>(column_position,
selectivity_percent));
+ context->_prepared = true;
+ context->_opened = true;
+ return context;
+}
+
+inline Block make_block(const std::vector<format::ColumnDefinition>& schema) {
+ Block block;
+ for (const auto& column : schema) {
+ block.insert({column.type->create_column(), column.type, column.name});
+ }
+ return block;
+}
+
+struct ReaderSession {
+ RuntimeState runtime_state {TQueryOptions(), TQueryGlobals()};
+ std::unique_ptr<format::parquet::ParquetReader> reader;
+ std::vector<format::ColumnDefinition> schema;
+ std::shared_ptr<format::FileScanRequest> request;
+};
+
+inline std::unique_ptr<ReaderSession> open_reader(const std::filesystem::path&
path,
+ const ReaderScenario&
scenario) {
+ auto session = std::make_unique<ReaderSession>();
+ auto properties = std::make_shared<io::FileSystemProperties>();
+ properties->system_type = TFileType::FILE_LOCAL;
+ auto description = std::make_unique<io::FileDescription>();
+ description->path = path.string();
+ description->file_size =
static_cast<int64_t>(std::filesystem::file_size(path));
+ description->range_start_offset = 0;
+ description->range_size = -1;
+ session->reader =
std::make_unique<format::parquet::ParquetReader>(properties, description,
+
nullptr, nullptr);
+ throw_if_error(session->reader->init(&session->runtime_state));
+ throw_if_error(session->reader->get_schema(&session->schema));
+
+ session->request = std::make_shared<format::FileScanRequest>();
+ format::FileScanRequestBuilder request_builder(session->request.get());
+ if (scenario.operation == ReaderOperation::FULL_SCAN) {
+ for (int column = 0; column < scenario.schema_width; ++column) {
+
throw_if_error(request_builder.add_non_predicate_column(format::LocalColumnId(column)));
+ }
+ } else if (scenario.operation == ReaderOperation::PREDICATE_SCAN) {
+ const auto predicate_id =
format::LocalColumnId(scenario.predicate_position);
+ throw_if_error(request_builder.add_predicate_column(predicate_id));
+ if (scenario.projection == Projection::PREDICATE_ONLY) {
+ session->request->predicate_only_columns.push_back(predicate_id);
+ } else {
+ const int payload = scenario.predicate_position == 0 ?
scenario.schema_width - 1 : 0;
+ throw_if_error(
+
request_builder.add_non_predicate_column(format::LocalColumnId(payload)));
+ }
+ const auto predicate_position =
session->request->local_positions.at(predicate_id).value();
+ session->request->conjuncts.push_back(
+ make_predicate(static_cast<int>(predicate_position),
scenario.selectivity_percent));
+ } else {
+
throw_if_error(request_builder.add_non_predicate_column(format::LocalColumnId(0)));
+ if (scenario.schema_width > 1) {
+
throw_if_error(request_builder.add_non_predicate_column(format::LocalColumnId(1)));
+ }
+ }
+
+ if (scenario.operation == ReaderOperation::LIMIT_1) {
+ session->reader->set_batch_size(1);
+ } else if (scenario.operation == ReaderOperation::LIMIT_1000) {
+ session->reader->set_batch_size(1000);
+ }
+ throw_if_error(session->reader->open(session->request));
+ return session;
+}
+
+inline size_t scan_reader(ReaderSession* session, const ReaderScenario&
scenario) {
+ size_t output_rows = 0;
+ bool eof = false;
+ const size_t limit = scenario.operation == ReaderOperation::LIMIT_1 ?
1
+ : scenario.operation == ReaderOperation::LIMIT_1000 ?
1000
+ :
0;
+ while (!eof) {
+ auto block = make_block(session->schema);
+ size_t rows = 0;
+ throw_if_error(session->reader->get_block(&block, &rows, &eof));
+ output_rows += rows;
+ benchmark::DoNotOptimize(block);
+ if (scenario.operation == ReaderOperation::OPEN_TO_FIRST_BLOCK) {
+ break;
+ }
+ if (limit != 0 && output_rows >= limit) {
+ break;
+ }
+ }
+ return output_rows;
+}
+
+inline size_t raw_rows_per_iteration(const ReaderScenario& scenario) {
+ if (scenario.operation == ReaderOperation::LIMIT_1) {
+ return 1;
+ }
+ if (scenario.operation == ReaderOperation::LIMIT_1000) {
+ return 1000;
+ }
+ if (scenario.operation == ReaderOperation::OPEN_TO_FIRST_BLOCK) {
+ return format::parquet::ParquetScanScheduler::DEFAULT_READ_BATCH_SIZE;
+ }
+ return READER_ROWS;
+}
+
+inline int projected_columns(const ReaderScenario& scenario) {
+ if (scenario.operation == ReaderOperation::FULL_SCAN) {
+ return scenario.schema_width;
+ }
+ if (scenario.operation == ReaderOperation::PREDICATE_SCAN &&
+ scenario.projection == Projection::PREDICATE_ONLY) {
+ return 1;
+ }
+ return std::min(2, scenario.schema_width);
+}
+
+inline void run_reader(benchmark::State& state, ReaderScenario scenario) {
+ std::filesystem::path fixture;
+ try {
+ fixture = ensure_fixture(scenario);
+ size_t selected_rows = 0;
+ for (auto _ : state) {
+ if (scenario.operation != ReaderOperation::OPEN_TO_FIRST_BLOCK) {
+ state.PauseTiming();
+ }
+ auto session = open_reader(fixture, scenario);
+ if (scenario.operation != ReaderOperation::OPEN_TO_FIRST_BLOCK) {
+ state.ResumeTiming();
+ }
+ selected_rows = scan_reader(session.get(), scenario);
+ state.PauseTiming();
+ throw_if_error(session->reader->close());
+ state.ResumeTiming();
+ benchmark::ClobberMemory();
+ }
+
+ const auto raw_rows = raw_rows_per_iteration(scenario);
+ state.SetItemsProcessed(static_cast<int64_t>(state.iterations() *
selected_rows));
+ state.SetBytesProcessed(static_cast<int64_t>(
+ state.iterations() * raw_rows * projected_columns(scenario) *
sizeof(int32_t)));
+ state.counters["raw_rows"] = static_cast<double>(raw_rows);
+ state.counters["selected_rows"] = static_cast<double>(selected_rows);
+ state.counters["fixture_bytes"] =
static_cast<double>(std::filesystem::file_size(fixture));
+ state.counters["ns/raw_row"] = benchmark::Counter(
+ static_cast<double>(raw_rows),
+ benchmark::Counter::kIsIterationInvariantRate |
benchmark::Counter::kInvert);
+ if (selected_rows > 0) {
+ state.counters["ns/selected_row"] = benchmark::Counter(
+ static_cast<double>(selected_rows),
+ benchmark::Counter::kIsIterationInvariantRate |
benchmark::Counter::kInvert);
+ }
+ } catch (const std::exception& error) {
+ state.SkipWithError(error.what());
+ }
+}
+
+inline bool register_reader_benchmarks() {
+ for (const auto& scenario : reader_scenarios()) {
+ std::string name =
+ "ParquetReader/" + to_string(scenario.operation) + "/" +
+ to_string(scenario.encoding) + "/null_" +
std::to_string(scenario.null_percent) +
+ "/" + to_string(scenario.null_pattern) + "/sel_" +
+ std::to_string(scenario.selectivity_percent) + "/" +
+ to_string(scenario.projection) + "/width_" +
std::to_string(scenario.schema_width) +
+ "/predicate_" + std::to_string(scenario.predicate_position);
+ benchmark::RegisterBenchmark(name.c_str(), [=](benchmark::State&
state) {
+ run_reader(state, scenario);
+ })->Unit(benchmark::kNanosecond);
+ }
+ return true;
+}
+
+inline const bool READER_BENCHMARKS_REGISTERED = register_reader_benchmarks();
+
+} // namespace reader_detail
+} // namespace doris::parquet_benchmark
diff --git a/be/benchmark/parquet/parquet_benchmark_scenarios.h
b/be/benchmark/parquet/parquet_benchmark_scenarios.h
new file mode 100644
index 00000000000..d420d9ccdd5
--- /dev/null
+++ b/be/benchmark/parquet/parquet_benchmark_scenarios.h
@@ -0,0 +1,250 @@
+// 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>
+#include <set>
+#include <string>
+#include <tuple>
+#include <vector>
+
+namespace doris::parquet_benchmark {
+
+enum class Encoding {
+ PLAIN,
+ DICTIONARY,
+ BYTE_STREAM_SPLIT,
+ DELTA_BINARY_PACKED,
+ DELTA_LENGTH_BYTE_ARRAY,
+ DELTA_BYTE_ARRAY
+};
+enum class ValueType { INT32, INT64, FLOAT, DOUBLE, BYTE_ARRAY,
FIXED_LEN_BYTE_ARRAY };
+enum class Pattern { CLUSTERED, ALTERNATING };
+enum class Projection { PREDICATE_ONLY, PREDICATE_PROJECTED };
+enum class ReaderOperation { OPEN_TO_FIRST_BLOCK, FULL_SCAN, PREDICATE_SCAN,
LIMIT_1, LIMIT_1000 };
+
+struct DecoderScenario {
+ Encoding encoding;
+ ValueType value_type;
+};
+
+struct ReaderScenario {
+ ReaderOperation operation;
+ Encoding encoding;
+ int null_percent;
+ Pattern null_pattern;
+ int selectivity_percent;
+ Projection projection;
+ int schema_width;
+ int predicate_position;
+};
+
+struct SelectionRange {
+ size_t first;
+ size_t count;
+};
+
+struct SelectionPlan {
+ size_t total_rows = 0;
+ size_t selected_rows = 0;
+ std::vector<SelectionRange> ranges;
+};
+
+inline std::vector<DecoderScenario> decoder_scenarios() {
+ return {
+ {Encoding::PLAIN, ValueType::INT32},
+ {Encoding::PLAIN, ValueType::INT64},
+ {Encoding::PLAIN, ValueType::FLOAT},
+ {Encoding::PLAIN, ValueType::DOUBLE},
+ {Encoding::PLAIN, ValueType::BYTE_ARRAY},
+ {Encoding::PLAIN, ValueType::FIXED_LEN_BYTE_ARRAY},
+ {Encoding::DICTIONARY, ValueType::INT32},
+ {Encoding::DICTIONARY, ValueType::INT64},
+ {Encoding::DICTIONARY, ValueType::FLOAT},
+ {Encoding::DICTIONARY, ValueType::DOUBLE},
+ {Encoding::DICTIONARY, ValueType::BYTE_ARRAY},
+ {Encoding::DICTIONARY, ValueType::FIXED_LEN_BYTE_ARRAY},
+ {Encoding::BYTE_STREAM_SPLIT, ValueType::FLOAT},
+ {Encoding::BYTE_STREAM_SPLIT, ValueType::DOUBLE},
+ {Encoding::BYTE_STREAM_SPLIT, ValueType::FIXED_LEN_BYTE_ARRAY},
+ {Encoding::DELTA_BINARY_PACKED, ValueType::INT32},
+ {Encoding::DELTA_BINARY_PACKED, ValueType::INT64},
+ {Encoding::DELTA_LENGTH_BYTE_ARRAY, ValueType::BYTE_ARRAY},
+ {Encoding::DELTA_BYTE_ARRAY, ValueType::BYTE_ARRAY},
+ };
+}
+
+inline std::vector<ReaderScenario> reader_scenarios() {
+ std::vector<ReaderScenario> scenarios;
+ std::set<std::tuple<ReaderOperation, Encoding, int, Pattern, int,
Projection, int, int>> seen;
+ const auto add = [&](ReaderScenario scenario) {
+ const auto key = std::make_tuple(scenario.operation, scenario.encoding,
+ scenario.null_percent,
scenario.null_pattern,
+ scenario.selectivity_percent,
scenario.projection,
+ scenario.schema_width,
scenario.predicate_position);
+ if (seen.insert(key).second) {
+ scenarios.push_back(scenario);
+ }
+ };
+
+ const ReaderScenario baseline {.operation = ReaderOperation::FULL_SCAN,
+ .encoding = Encoding::PLAIN,
+ .null_percent = 10,
+ .null_pattern = Pattern::ALTERNATING,
+ .selectivity_percent = 10,
+ .projection =
Projection::PREDICATE_PROJECTED,
+ .schema_width = 32,
+ .predicate_position = 0};
+ for (const auto operation :
+ {ReaderOperation::OPEN_TO_FIRST_BLOCK, ReaderOperation::FULL_SCAN,
+ ReaderOperation::PREDICATE_SCAN, ReaderOperation::LIMIT_1,
ReaderOperation::LIMIT_1000}) {
+ auto scenario = baseline;
+ scenario.operation = operation;
+ add(scenario);
+ }
+ for (const auto encoding : {Encoding::PLAIN, Encoding::DICTIONARY,
Encoding::BYTE_STREAM_SPLIT,
+ Encoding::DELTA_BINARY_PACKED}) {
+ auto scenario = baseline;
+ scenario.encoding = encoding;
+ add(scenario);
+ scenario.operation = ReaderOperation::PREDICATE_SCAN;
+ add(scenario);
+ }
+ for (const int width : {4, 32, 128, 512}) {
+ for (const int predicate_position : {0, width - 1}) {
+ auto scenario = baseline;
+ scenario.operation = ReaderOperation::PREDICATE_SCAN;
+ scenario.schema_width = width;
+ scenario.predicate_position = predicate_position;
+ add(scenario);
+ }
+ }
+ 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}) {
+ for (const auto projection :
+ {Projection::PREDICATE_ONLY,
Projection::PREDICATE_PROJECTED}) {
+ auto scenario = baseline;
+ scenario.operation = ReaderOperation::PREDICATE_SCAN;
+ scenario.null_percent = null_percent;
+ scenario.null_pattern = pattern;
+ scenario.selectivity_percent = selectivity;
+ scenario.projection = projection;
+ add(scenario);
+ }
+ }
+ }
+ }
+ return scenarios;
+}
+
+inline SelectionPlan make_selection_plan(size_t total_rows, int
selectivity_percent,
+ Pattern pattern) {
+ SelectionPlan plan {.total_rows = total_rows, .selected_rows = 0, .ranges
= {}};
+ if (total_rows == 0 || selectivity_percent <= 0) {
+ return plan;
+ }
+ if (selectivity_percent >= 100) {
+ plan.selected_rows = total_rows;
+ plan.ranges.push_back({.first = 0, .count = total_rows});
+ return plan;
+ }
+ plan.selected_rows = total_rows * static_cast<size_t>(selectivity_percent)
/ 100;
+ if (plan.selected_rows == 0) {
+ plan.selected_rows = 1;
+ }
+ if (pattern == Pattern::CLUSTERED) {
+ plan.ranges.push_back({.first = 0, .count = plan.selected_rows});
+ return plan;
+ }
+
+ // Evenly spaced rows deliberately maximize the number of physical ranges.
This is the
+ // adversarial sparse shape that exposes per-run decoder and cursor
overhead.
+ for (size_t selected = 0; selected < plan.selected_rows; ++selected) {
+ const size_t row = selected * total_rows / plan.selected_rows;
+ if (!plan.ranges.empty() && plan.ranges.back().first +
plan.ranges.back().count == row) {
+ ++plan.ranges.back().count;
+ } else {
+ plan.ranges.push_back({.first = row, .count = 1});
+ }
+ }
+ return plan;
+}
+
+inline std::string to_string(Encoding value) {
+ switch (value) {
+ case Encoding::PLAIN:
+ return "plain";
+ case Encoding::DICTIONARY:
+ return "dictionary";
+ case Encoding::BYTE_STREAM_SPLIT:
+ return "byte_stream_split";
+ case Encoding::DELTA_BINARY_PACKED:
+ return "delta_binary_packed";
+ case Encoding::DELTA_LENGTH_BYTE_ARRAY:
+ return "delta_length_byte_array";
+ case Encoding::DELTA_BYTE_ARRAY:
+ return "delta_byte_array";
+ }
+ return "unknown";
+}
+
+inline std::string to_string(ValueType value) {
+ switch (value) {
+ case ValueType::INT32:
+ return "int32";
+ case ValueType::INT64:
+ return "int64";
+ case ValueType::FLOAT:
+ return "float";
+ case ValueType::DOUBLE:
+ return "double";
+ case ValueType::BYTE_ARRAY:
+ return "byte_array";
+ case ValueType::FIXED_LEN_BYTE_ARRAY:
+ return "fixed_len_byte_array";
+ }
+ return "unknown";
+}
+
+inline std::string to_string(Pattern value) {
+ return value == Pattern::CLUSTERED ? "clustered" : "alternating";
+}
+
+inline std::string to_string(Projection value) {
+ return value == Projection::PREDICATE_ONLY ? "predicate_only" :
"predicate_projected";
+}
+
+inline std::string to_string(ReaderOperation value) {
+ switch (value) {
+ case ReaderOperation::OPEN_TO_FIRST_BLOCK:
+ return "open_to_first_block";
+ case ReaderOperation::FULL_SCAN:
+ return "full_scan";
+ case ReaderOperation::PREDICATE_SCAN:
+ return "predicate_scan";
+ case ReaderOperation::LIMIT_1:
+ return "limit_1";
+ case ReaderOperation::LIMIT_1000:
+ return "limit_1000";
+ }
+ return "unknown";
+}
+
+} // namespace doris::parquet_benchmark
diff --git a/be/test/format_v2/parquet/parquet_benchmark_scenarios_test.cpp
b/be/test/format_v2/parquet/parquet_benchmark_scenarios_test.cpp
new file mode 100644
index 00000000000..c62d3566e15
--- /dev/null
+++ b/be/test/format_v2/parquet/parquet_benchmark_scenarios_test.cpp
@@ -0,0 +1,129 @@
+// 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 "../../../benchmark/parquet/parquet_benchmark_scenarios.h"
+
+#include <gtest/gtest.h>
+
+#include <algorithm>
+#include <array>
+#include <set>
+#include <tuple>
+
+namespace doris::parquet_benchmark {
+namespace {
+
+TEST(ParquetBenchmarkScenariosTest,
DecoderMatrixCoversNativeEncodingAndTypeFamilies) {
+ const auto scenarios = decoder_scenarios();
+ const std::set<std::pair<Encoding, ValueType>> actual = [&] {
+ std::set<std::pair<Encoding, ValueType>> values;
+ for (const auto& scenario : scenarios) {
+ values.emplace(scenario.encoding, scenario.value_type);
+ }
+ return values;
+ }();
+
+ const std::set<std::pair<Encoding, ValueType>> expected {
+ {Encoding::PLAIN, ValueType::INT32},
+ {Encoding::PLAIN, ValueType::INT64},
+ {Encoding::PLAIN, ValueType::FLOAT},
+ {Encoding::PLAIN, ValueType::DOUBLE},
+ {Encoding::PLAIN, ValueType::BYTE_ARRAY},
+ {Encoding::PLAIN, ValueType::FIXED_LEN_BYTE_ARRAY},
+ {Encoding::DICTIONARY, ValueType::INT32},
+ {Encoding::DICTIONARY, ValueType::INT64},
+ {Encoding::DICTIONARY, ValueType::FLOAT},
+ {Encoding::DICTIONARY, ValueType::DOUBLE},
+ {Encoding::DICTIONARY, ValueType::BYTE_ARRAY},
+ {Encoding::DICTIONARY, ValueType::FIXED_LEN_BYTE_ARRAY},
+ {Encoding::BYTE_STREAM_SPLIT, ValueType::FLOAT},
+ {Encoding::BYTE_STREAM_SPLIT, ValueType::DOUBLE},
+ {Encoding::BYTE_STREAM_SPLIT, ValueType::FIXED_LEN_BYTE_ARRAY},
+ {Encoding::DELTA_BINARY_PACKED, ValueType::INT32},
+ {Encoding::DELTA_BINARY_PACKED, ValueType::INT64},
+ {Encoding::DELTA_LENGTH_BYTE_ARRAY, ValueType::BYTE_ARRAY},
+ {Encoding::DELTA_BYTE_ARRAY, ValueType::BYTE_ARRAY},
+ };
+ EXPECT_EQ(actual, expected);
+}
+
+TEST(ParquetBenchmarkScenariosTest,
ReaderMatrixCoversNullableSparseAndProjectionAxes) {
+ const auto scenarios = reader_scenarios();
+ 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}) {
+ for (const auto projection :
+ {Projection::PREDICATE_ONLY,
Projection::PREDICATE_PROJECTED}) {
+ EXPECT_TRUE(std::ranges::any_of(scenarios, [&](const
ReaderScenario& scenario) {
+ return scenario.operation ==
ReaderOperation::PREDICATE_SCAN &&
+ scenario.encoding == Encoding::PLAIN &&
+ scenario.null_percent == null_percent &&
+ scenario.null_pattern == pattern &&
+ scenario.selectivity_percent == selectivity &&
+ scenario.projection == projection &&
scenario.schema_width == 32;
+ })) << "missing nullable sparse axis combination";
+ }
+ }
+ }
+ }
+}
+
+TEST(ParquetBenchmarkScenariosTest,
ReaderMatrixCoversOperationsEncodingsAndSchemaWidths) {
+ const auto scenarios = reader_scenarios();
+ for (const auto operation :
+ {ReaderOperation::OPEN_TO_FIRST_BLOCK, ReaderOperation::FULL_SCAN,
+ ReaderOperation::PREDICATE_SCAN, ReaderOperation::LIMIT_1,
ReaderOperation::LIMIT_1000}) {
+ EXPECT_TRUE(std::ranges::any_of(scenarios, [&](const ReaderScenario&
scenario) {
+ return scenario.operation == operation;
+ }));
+ }
+ for (const auto encoding : {Encoding::PLAIN, Encoding::DICTIONARY,
Encoding::BYTE_STREAM_SPLIT,
+ Encoding::DELTA_BINARY_PACKED}) {
+ for (const auto operation : {ReaderOperation::FULL_SCAN,
ReaderOperation::PREDICATE_SCAN}) {
+ EXPECT_TRUE(std::ranges::any_of(scenarios, [&](const
ReaderScenario& scenario) {
+ return scenario.encoding == encoding && scenario.operation ==
operation;
+ }));
+ }
+ }
+ for (const int width : {4, 32, 128, 512}) {
+ for (const int predicate_position : {0, width - 1}) {
+ EXPECT_TRUE(std::ranges::any_of(scenarios, [&](const
ReaderScenario& scenario) {
+ return scenario.schema_width == width &&
+ scenario.predicate_position == predicate_position;
+ }));
+ }
+ }
+}
+
+TEST(ParquetBenchmarkScenariosTest,
SelectionPlanDistinguishesClusteredAndSparseRuns) {
+ const auto clustered = make_selection_plan(1000, 10, Pattern::CLUSTERED);
+ EXPECT_EQ(clustered.total_rows, 1000);
+ EXPECT_EQ(clustered.selected_rows, 100);
+ ASSERT_EQ(clustered.ranges.size(), 1);
+ EXPECT_EQ(clustered.ranges.front().count, 100);
+
+ const auto alternating = make_selection_plan(1000, 10,
Pattern::ALTERNATING);
+ EXPECT_EQ(alternating.total_rows, 1000);
+ EXPECT_EQ(alternating.selected_rows, 100);
+ EXPECT_EQ(alternating.ranges.size(), 100);
+
+ EXPECT_EQ(make_selection_plan(1000, 0,
Pattern::ALTERNATING).ranges.size(), 0);
+ EXPECT_EQ(make_selection_plan(1000, 100,
Pattern::ALTERNATING).ranges.size(), 1);
+}
+
+} // namespace
+} // namespace doris::parquet_benchmark
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
