This is an automated email from the ASF dual-hosted git repository.
eldenmoon 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 243df79a0c0 [test](be) Extend Variant V1/V2 segment benchmarks (#66668)
243df79a0c0 is described below
commit 243df79a0c0579bf4469517a66888dd635a35df3
Author: lihangyu <[email protected]>
AuthorDate: Thu Aug 20 15:39:08 2026 +0800
[test](be) Extend Variant V1/V2 segment benchmarks (#66668)
ColumnVariantV2 needs a reproducible component benchmark that exercises
real segment write/read paths, scales beyond one segment, distinguishes
materialized and sparse exact-path reads, and measures the current
production compaction pipeline without inventing a benchmark-only V2
compaction mode.
### What is changed?
- Keep 1,000,000 rows as the checked-in default and add
environment-controlled total rows and rows per segment. Input is
generated in 4K batches, while parse/append/finalize remain in the timed
region.
- Add multi-segment ingestion for `Sparse16`, `Doc16`, and `Full`
layouts.
- Add exact-path read cases corresponding to the storage-scan portion of
a literal `SELECT v['path']` after subpath pruning:
- `Materialized100pct`: materialized `LEAF` route.
- `Sparse6pct`: sparse-bucket `BINARY_EXTRACT` route with one hit per 17
rows.
- Add a scan-and-rewrite pipeline benchmark. It queries key plus whole
Variant from the same
prevalidated V1-written source Segment as `ColumnVariant` or
`ColumnVariantV2`, then passes that
queried Block directly to a new `SegmentWriter` and reports every
pipeline stage separately.
- Add a current-production cumulative-compaction benchmark with ten
overlapping rowsets written by Variant V2. It calls public
`CumulativeCompaction::execute_compact()`, disables and verifies the
ordered-link shortcut, and validates the physically rewritten output.
- Keep correctness checks outside timing: full row/layout/route checks,
bounded whole-value canonical samples, and full exact-path checksums/hit
counts.
- Do not use whole-column read performance as a gate in this round.
The compaction benchmark intentionally does **not** label the current
compactor as V2. `TabletColumn::variant_is_v2` is a transient read
marker and is not persisted through schema protobuf copies, so the
benchmark records `persisted_compaction_variant_v2=0` instead of
creating a misleading V1/V2 pair.
### Release benchmark evidence
- Default 1M ingestion, five samples per representation:
| Layout | V1 median (ms) | V2 median (ms) | V2/V1 |
|---|---:|---:|---:|
| Sparse16 | 9,212.349 | 8,202.836 | 0.8904 |
| Doc16 | 7,917.702 | 8,286.466 | 1.0466 |
| Full | 10,231.216 | 9,195.444 | 0.8988 |
- Final exact-path 1M, five samples per representation:
| Route | V1 median (ms) | V2 median (ms) | V2/V1 |
|---|---:|---:|---:|
| Materialized `LEAF` | 3.630 | 2.403 | 0.6619 |
| Sparse `BINARY_EXTRACT` | 28.254 | 27.006 | 0.9558 |
#### Scan and rewrite: 1M rows
The source is the same V1-written physical Segment for both sides. The
only switched dimension is
the requested query representation and therefore the column handed to
the destination writer.
Every group contains five V1 and five V2 samples in fixed ABBA order.
Lower is better.
| Layout | Representation | Total median (ms) | Read median (ms) |
Append median (ms) | Finalize median (ms) | Total CV |
|---|---|---:|---:|---:|---:|---:|
| Sparse16 | V1 | 8,429.347 | 3,347.014 | 1,580.744 | 3,501.255 | 0.554%
|
| Sparse16 | V2 | 11,275.605 | 5,869.060 | 2,005.063 | 3,401.644 |
0.543% |
| Doc16 | V1 | 7,035.376 | 2,719.646 | 699.392 | 3,610.837 | 0.238% |
| Doc16 | V2 | 11,568.249 | 6,106.403 | 1,977.956 | 3,485.382 | 0.095% |
| Full | V1 | 8,023.795 | 2,010.926 | 1,166.440 | 4,804.266 | 0.212% |
| Full | V2 | 13,154.337 | 4,567.923 | 4,201.698 | 4,400.651 | 0.799% |
V2/V1 total latency is `1.3377` for Sparse16, `1.6443` for Doc16, and
`1.6394`
for Full. The gap is in the query/read and append stages; destination
finalize is comparable or
slightly faster for V2 (`0.92x`-`0.97x`). Destination segment size and
physical layout match for V1
and V2 in every layout.
---
be/benchmark/benchmark_main.cpp | 1 +
be/benchmark/benchmark_variant_segment.hpp | 2037 +++++++++++++++++++++++++
be/benchmark/run_variant_segment_benchmark.sh | 124 ++
be/src/exec/common/variant_util.cpp | 4 -
be/src/exec/common/variant_util.h | 4 +
be/test/storage/variant/variant_util_test.cpp | 33 +
6 files changed, 2199 insertions(+), 4 deletions(-)
diff --git a/be/benchmark/benchmark_main.cpp b/be/benchmark/benchmark_main.cpp
index aed5c6eb84e..30be63926a0 100644
--- a/be/benchmark/benchmark_main.cpp
+++ b/be/benchmark/benchmark_main.cpp
@@ -31,6 +31,7 @@
#include "benchmark_fmod.hpp"
#include "benchmark_hll_merge.hpp"
#include "benchmark_json_extract.hpp"
+#include "benchmark_variant_segment.hpp"
#include "benchmark_zone_map_index.hpp"
#include "binary_cast_benchmark.hpp"
#include "common/config.h"
diff --git a/be/benchmark/benchmark_variant_segment.hpp
b/be/benchmark/benchmark_variant_segment.hpp
new file mode 100644
index 00000000000..b6721ec5faf
--- /dev/null
+++ b/be/benchmark/benchmark_variant_segment.hpp
@@ -0,0 +1,2037 @@
+// 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 <sys/stat.h>
+#include <unistd.h>
+
+#include <algorithm>
+#include <array>
+#include <cerrno>
+#include <charconv>
+#include <chrono>
+#include <cstdint>
+#include <cstdlib>
+#include <cstring>
+#include <limits>
+#include <memory>
+#include <string>
+#include <string_view>
+#include <system_error>
+#include <utility>
+#include <vector>
+
+#include "common/status.h"
+#include "core/block/block.h"
+#include "core/column/column_nullable.h"
+#include "core/column/column_string.h"
+#include "core/column/column_variant.h"
+#include "core/column/column_vector.h"
+#include "core/column/variant_v2/column_variant_v2.h"
+#include "core/data_type/data_type_number.h"
+#include "core/data_type/data_type_variant.h"
+#include "core/data_type/data_type_variant_v2.h"
+#include "exec/common/variant_util.h"
+#include "exprs/function/parse/variant_string_parse.h"
+#include "io/fs/file_writer.h"
+#include "io/fs/local_file_system.h"
+#include "runtime/exec_env.h"
+#include "runtime/memory/cache_manager.h"
+#include "storage/cache/page_cache.h"
+#include "storage/compaction/cumulative_compaction.h"
+#include "storage/data_dir.h"
+#include "storage/index/index_writer.h"
+#include "storage/olap_common.h"
+#include "storage/options.h"
+#include "storage/rowset/beta_rowset.h"
+#include "storage/rowset/rowset_factory.h"
+#include "storage/rowset/rowset_reader.h"
+#include "storage/rowset/rowset_reader_context.h"
+#include "storage/rowset/rowset_writer.h"
+#include "storage/rowset/rowset_writer_context.h"
+#include "storage/schema.h"
+#include "storage/segment/column_meta_accessor.h"
+#include "storage/segment/segment.h"
+#include "storage/segment/segment_loader.h"
+#include "storage/segment/segment_writer.h"
+#include "storage/storage_engine.h"
+#include "storage/tablet/tablet.h"
+#include "storage/tablet/tablet_column_object_pool.h"
+#include "storage/tablet/tablet_meta.h"
+#include "storage/tablet/tablet_schema.h"
+#include "storage/tablet/tablet_schema_cache.h"
+
+namespace doris::variant_segment_benchmark {
+namespace {
+
+// P0 workload: 200 top-level BIGINT paths, with one hot path and 29
additional paths per row.
+// Ingest cases parse the same canonical JSON before SegmentWriter. Read cases
use the same
+// V1-written physical segment and only switch the requested V1/V2 output
representation. Input
+// generation, buffer destruction, warmup, checksum, physical-layout
validation, and route
+// validation are paused. Scan-and-rewrite cases query key plus whole Variant
from that prevalidated
+// source, then time scan/writer initialization, read, append, and destination
finalize; destination
+// validation is paused. Ingest rotates SegmentWriter at
+// DORIS_VARIANT_BENCHMARK_ROWS_PER_SEGMENT; read cases retain their
historical single-segment
+// semantics.
+constexpr uint32_t DEFAULT_ROWS = 1'000'000;
+constexpr uint32_t DEFAULT_ROWS_PER_SEGMENT = 1'000'000;
+constexpr uint32_t BATCH_ROWS = 4'096;
+constexpr uint32_t WHOLE_VALIDATION_STRIDE = BATCH_ROWS;
+constexpr uint32_t COMPACTION_INPUT_ROWSETS = 10;
+constexpr uint32_t CANDIDATE_PATHS = 200;
+constexpr uint32_t COLD_PATHS = CANDIDATE_PATHS - 2;
+constexpr uint32_t FIELDS_PER_ROW = 30;
+constexpr uint32_t SPARSE_HIT_PERIOD = 17;
+constexpr int32_t KEY_UID = 0;
+constexpr int32_t ROOT_UID = 1;
+constexpr uint32_t BUCKETS = 16;
+constexpr std::string_view ROOT_NAME = "v";
+constexpr std::string_view HOT_PATH = "hot";
+constexpr std::string_view SPARSE_PATH = "sparse_target";
+constexpr std::string_view MISSING_PATH = "global_missing";
+constexpr uint64_t FNV_OFFSET = 1469598103934665603ULL;
+constexpr uint64_t FNV_PRIME = 1099511628211ULL;
+
+enum class VariantVersion : uint8_t { V1, V2 };
+enum class VariantLayout : uint8_t { SPARSE16, DOC16, FULL };
+enum class ReadTarget : uint8_t { WHOLE, MATERIALIZED, SPARSE, MISSING };
+
+struct LayoutConfig {
+ std::string_view name;
+ int32_t max_subcolumns;
+ bool doc_mode;
+};
+
+struct LayoutCounts {
+ uint32_t materialized = 0;
+ uint32_t sparse = 0;
+ uint32_t doc = 0;
+};
+
+void count_layout_columns(const ColumnMetaPB& meta, LayoutCounts* counts) {
+ DORIS_CHECK(counts != nullptr);
+ if (meta.has_column_path_info()) {
+ PathInData path;
+ path.from_protobuf(meta.column_path_info());
+ const std::string relative = path.copy_pop_front().get_path();
+ if (!relative.empty()) {
+ if (relative.find(SPARSE_COLUMN_PATH) != std::string::npos) {
+ ++counts->sparse;
+ } else if (relative.find(DOC_VALUE_COLUMN_PATH) !=
std::string::npos) {
+ ++counts->doc;
+ } else {
+ ++counts->materialized;
+ }
+ }
+ }
+ for (const ColumnMetaPB& child : meta.children_columns()) {
+ count_layout_columns(child, counts);
+ }
+}
+
+struct WriteResult {
+ uint64_t segment_bytes = 0;
+ uint64_t index_bytes = 0;
+ int64_t init_ns = 0;
+ int64_t parse_ns = 0;
+ int64_t append_ns = 0;
+ int64_t finalize_ns = 0;
+};
+
+struct PreparedSegment {
+ TabletSchemaSPtr schema;
+ segment_v2::SegmentSharedPtr segment;
+ std::string path;
+ uint64_t segment_bytes = 0;
+ LayoutCounts counts;
+};
+
+struct PreparedScan {
+ PreparedSegment* fixture = nullptr;
+ TabletSchemaSPtr query_schema;
+ SchemaSPtr scan_schema;
+ ColumnId output_column_id = 0;
+ ReadTarget target = ReadTarget::WHOLE;
+};
+
+struct ScanResult {
+ uint64_t checksum = FNV_OFFSET;
+ uint64_t output_bytes = 0;
+ uint32_t rows = 0;
+ uint32_t hits = 0;
+ OlapReaderStatistics statistics;
+};
+
+struct RewriteResult {
+ uint64_t source_segment_bytes = 0;
+ uint64_t read_output_bytes = 0;
+ uint64_t destination_segment_bytes = 0;
+ uint64_t destination_index_bytes = 0;
+ uint32_t rows = 0;
+ int64_t scan_init_ns = 0;
+ int64_t writer_init_ns = 0;
+ int64_t read_ns = 0;
+ int64_t append_ns = 0;
+ int64_t finalize_ns = 0;
+ OlapReaderStatistics statistics;
+};
+
+struct CompactionResult {
+ uint64_t input_json_bytes = 0;
+ uint64_t input_disk_bytes = 0;
+ uint64_t output_disk_bytes = 0;
+ uint32_t output_segments = 0;
+ LayoutCounts output_layout;
+};
+
+std::string_view version_name(VariantVersion version) {
+ return version == VariantVersion::V1 ? "V1" : "V2";
+}
+
+LayoutConfig layout_config(VariantLayout layout) {
+ switch (layout) {
+ case VariantLayout::SPARSE16:
+ return {.name = "Sparse16", .max_subcolumns = 1, .doc_mode = false};
+ case VariantLayout::DOC16:
+ return {.name = "Doc16", .max_subcolumns = 1, .doc_mode = true};
+ case VariantLayout::FULL:
+ return {.name = "Full", .max_subcolumns = 0, .doc_mode = false};
+ }
+ __builtin_unreachable();
+}
+
+Status validate_sparse16_layout(const LayoutCounts& counts) {
+ if (counts.materialized != 1 || counts.sparse != BUCKETS || counts.doc !=
0) {
+ return Status::InternalError("Sparse16 layout is materialized={},
sparse={}, doc={}",
+ counts.materialized, counts.sparse,
counts.doc);
+ }
+ return Status::OK();
+}
+
+void ensure_variant_compaction_runtime() {
+ ExecEnv* env = ExecEnv::GetInstance();
+ if (env->get_cache_manager() == nullptr) {
+ env->set_cache_manager(CacheManager::create_global_instance());
+ }
+ if (env->get_storage_page_cache() == nullptr) {
+ constexpr size_t CACHE_CAPACITY = 512UL << 20;
+
env->set_storage_page_cache(StoragePageCache::create_global_cache(CACHE_CAPACITY,
10, 0));
+ }
+ if (env->segment_loader() == nullptr) {
+ static const std::unique_ptr<SegmentLoader> loader =
+ std::make_unique<SegmentLoader>(512UL << 20, 4'096);
+ env->set_segment_loader(loader.get());
+ }
+ if (env->get_tablet_schema_cache() == nullptr) {
+
env->set_tablet_schema_cache(TabletSchemaCache::create_global_schema_cache(
+ config::tablet_schema_cache_capacity));
+ }
+ if (env->get_tablet_column_object_pool() == nullptr) {
+
env->set_tablet_column_object_pool(TabletColumnObjectPool::create_global_column_cache(
+ config::tablet_schema_cache_capacity));
+ }
+}
+
+std::string_view target_name(ReadTarget target) {
+ switch (target) {
+ case ReadTarget::WHOLE:
+ return "Whole";
+ case ReadTarget::MATERIALIZED:
+ return "Materialized100pct";
+ case ReadTarget::SPARSE:
+ return "Sparse6pct";
+ case ReadTarget::MISSING:
+ return "GlobalMiss";
+ }
+ __builtin_unreachable();
+}
+
+uint32_t configured_rows() {
+ static const uint32_t rows = [] {
+ const char* value = std::getenv("DORIS_VARIANT_BENCHMARK_ROWS");
+ if (value == nullptr) {
+ return DEFAULT_ROWS;
+ }
+ uint64_t parsed = 0;
+ const std::string_view text(value);
+ const auto [end, error] = std::from_chars(text.data(), text.data() +
text.size(), parsed);
+ DORIS_CHECK(error == std::errc {} && end == text.data() + text.size());
+ DORIS_CHECK_GT(parsed, 0);
+ DORIS_CHECK_LE(parsed, std::numeric_limits<uint32_t>::max());
+ return static_cast<uint32_t>(parsed);
+ }();
+ return rows;
+}
+
+uint32_t configured_rows_per_segment() {
+ static const uint32_t rows = [] {
+ const char* value =
std::getenv("DORIS_VARIANT_BENCHMARK_ROWS_PER_SEGMENT");
+ if (value == nullptr) {
+ return DEFAULT_ROWS_PER_SEGMENT;
+ }
+ uint64_t parsed = 0;
+ const std::string_view text(value);
+ const auto [end, error] = std::from_chars(text.data(), text.data() +
text.size(), parsed);
+ DORIS_CHECK(error == std::errc {} && end == text.data() + text.size());
+ DORIS_CHECK_GT(parsed, 0);
+ DORIS_CHECK_LE(parsed, std::numeric_limits<uint32_t>::max());
+ return static_cast<uint32_t>(parsed);
+ }();
+ return rows;
+}
+
+std::string benchmark_root() {
+ const char* value = std::getenv("DORIS_VARIANT_BENCHMARK_ROOT");
+ return value == nullptr ? "/tmp" : value;
+}
+
+uint64_t update_checksum(uint64_t checksum, std::string_view value) {
+ for (const unsigned char byte : value) {
+ checksum ^= byte;
+ checksum *= FNV_PRIME;
+ }
+ checksum ^= 0xff;
+ checksum *= FNV_PRIME;
+ return checksum;
+}
+
+int64_t elapsed_ns(std::chrono::steady_clock::time_point start) {
+ return
std::chrono::duration_cast<std::chrono::nanoseconds>(std::chrono::steady_clock::now()
-
+ start)
+ .count();
+}
+
+class VariantSegmentBenchmarkData {
+public:
+ static VariantSegmentBenchmarkData& instance() {
+ static VariantSegmentBenchmarkData data;
+ return data;
+ }
+
+ const Status& status() const { return _status; }
+ uint32_t rows() const { return _rows; }
+ uint32_t rows_per_segment() const { return _rows_per_segment; }
+ uint32_t expected_hits(ReadTarget target) const {
+ if (target == ReadTarget::MATERIALIZED) {
+ return _rows;
+ }
+ if (target == ReadTarget::SPARSE) {
+ return (_rows + SPARSE_HIT_PERIOD - 1) / SPARSE_HIT_PERIOD;
+ }
+ return 0;
+ }
+ uint32_t segment_count() const {
+ return static_cast<uint32_t>((static_cast<uint64_t>(_rows) +
_rows_per_segment - 1) /
+ _rows_per_segment);
+ }
+ uint32_t whole_validation_sample_rows() const {
+ uint64_t samples = 0;
+ for (uint32_t segment_id = 0; segment_id < segment_count();
++segment_id) {
+ const uint32_t rows = rows_in_segment(segment_id);
+ samples += (static_cast<uint64_t>(rows) + WHOLE_VALIDATION_STRIDE
- 1) /
+ WHOLE_VALIDATION_STRIDE;
+ if ((rows - 1) % WHOLE_VALIDATION_STRIDE != 0) {
+ ++samples;
+ }
+ }
+ DORIS_CHECK_LE(samples, std::numeric_limits<uint32_t>::max());
+ return static_cast<uint32_t>(samples);
+ }
+ uint64_t input_bytes() const { return _input_bytes; }
+ std::string measured_segment_path(VariantLayout layout, VariantVersion
version,
+ std::string benchmark_name) const {
+ std::replace(benchmark_name.begin(), benchmark_name.end(), '/', '_');
+ return _directory + "/measured_" +
std::string(layout_config(layout).name) + "_" +
+ std::string(version_name(version)) + "_" + benchmark_name +
".dat";
+ }
+ Status delete_segment_files(const std::string& base_path) const {
+ for (uint32_t segment_id = 0; segment_id < segment_count();
++segment_id) {
+ RETURN_IF_ERROR(io::global_local_filesystem()->delete_file(
+ segment_path(base_path, segment_id)));
+ }
+ return Status::OK();
+ }
+ ParseConfig v1_parse_config(VariantLayout layout) const {
+ const TabletSchema& schema = *_schemas[static_cast<size_t>(layout)];
+ const int32_t root_index = schema.field_index(ROOT_UID);
+ DORIS_CHECK_GE(root_index, 0);
+ ParseConfig parse_config;
+ parse_config.deprecated_enable_flatten_nested =
schema.deprecated_variant_flatten_nested();
+ parse_config.check_duplicate_json_path =
config::variant_enable_duplicate_json_path_check;
+ parse_config.parse_to =
variant_util::select_storage_variant_parse_target(
+ schema.column(root_index), parse_config);
+ return parse_config;
+ }
+ Status ensure_writer_warmup(VariantLayout layout, VariantVersion version,
+ PreparedSegment** prepared) {
+ if (prepared != nullptr && segment_count() != 1) {
+ return Status::InternalError(
+ "Variant read benchmark requires one segment, but rows={}
and "
+ "rows_per_segment={} produce {} segments",
+ _rows, _rows_per_segment, segment_count());
+ }
+ const auto layout_index = static_cast<size_t>(layout);
+ const auto version_index = static_cast<size_t>(version);
+ if (_writer_attempted[layout_index][version_index]) {
+ if (!_writer_errors[layout_index][version_index].empty()) {
+ return
Status::InternalError(_writer_errors[layout_index][version_index]);
+ }
+ if (prepared != nullptr) {
+ *prepared = &_writer_segments[layout_index][version_index];
+ }
+ return Status::OK();
+ }
+ _writer_attempted[layout_index][version_index] = true;
+
+ PreparedSegment result;
+ const std::string path = _directory + "/warm_" +
std::string(layout_config(layout).name) +
+ "_" + std::string(version_name(version)) +
".dat";
+ RETURN_IF_ERROR(delete_segment_files(path));
+ WriteResult write_result;
+ std::vector<uint64_t> segment_bytes;
+ Status status =
+ write_segments(layout, version, path, nullptr, &write_result,
&segment_bytes);
+ if (status.ok()) {
+ LayoutCounts counts;
+ status = validate_written_segments(layout, version, path,
segment_bytes, &counts,
+ segment_count() == 1 ? &result
: nullptr);
+ }
+ if (!status.ok()) {
+ _writer_errors[layout_index][version_index] = status.to_string();
+ return status;
+ }
+ if (segment_count() == 1) {
+ _writer_segments[layout_index][version_index] = std::move(result);
+ }
+ if (prepared != nullptr) {
+ *prepared = &_writer_segments[layout_index][version_index];
+ }
+ return Status::OK();
+ }
+
+ Status ensure_read_validation(VariantLayout layout, ReadTarget target,
+ PreparedSegment** fixture) {
+ const auto layout_index = static_cast<size_t>(layout);
+ const auto target_index = static_cast<size_t>(target);
+ PreparedSegment* segment = nullptr;
+ RETURN_IF_ERROR(ensure_writer_warmup(layout, VariantVersion::V1,
&segment));
+
+ if (_read_attempted[layout_index][target_index]) {
+ if (!_read_errors[layout_index][target_index].empty()) {
+ return
Status::InternalError(_read_errors[layout_index][target_index]);
+ }
+ *fixture = segment;
+ return Status::OK();
+ }
+ _read_attempted[layout_index][target_index] = true;
+
+ PreparedScan v1_scan;
+ PreparedScan v2_scan;
+ Status status = prepare_scan(segment, target, VariantVersion::V1,
&v1_scan);
+ if (status.ok()) {
+ status = prepare_scan(segment, target, VariantVersion::V2,
&v2_scan);
+ }
+ ScanResult v1_result;
+ ScanResult v2_result;
+ const bool sample_whole = target == ReadTarget::WHOLE;
+ if (status.ok()) {
+ status = scan_segment(v1_scan, true, _rows, &v1_result,
sample_whole);
+ }
+ if (status.ok()) {
+ status = scan_segment(v2_scan, true, _rows, &v2_result,
sample_whole);
+ }
+ if (status.ok() && (v1_result.rows != v2_result.rows || v1_result.hits
!= v2_result.hits)) {
+ status = Status::InternalError("{} {} V1/V2 mismatch: rows {}/{},
hits {}/{}",
+ layout_config(layout).name,
target_name(target),
+ v1_result.rows, v2_result.rows,
v1_result.hits,
+ v2_result.hits);
+ }
+ const auto expected = expected_checksum(target);
+ if (status.ok() && !sample_whole &&
+ (v1_result.checksum != expected || v2_result.checksum !=
expected)) {
+ status = Status::InternalError("{} {} checksum mismatch: V1 {}, V2
{}, input {}",
+ layout_config(layout).name,
target_name(target),
+ v1_result.checksum,
v2_result.checksum, expected);
+ }
+ if (status.ok()) {
+ status = validate_hit_count(target, v1_result.hits);
+ }
+ if (status.ok()) {
+ status = validate_route(target, v1_result.statistics);
+ }
+ if (status.ok()) {
+ status = validate_route(target, v2_result.statistics);
+ }
+ if (!status.ok()) {
+ _read_errors[layout_index][target_index] = status.to_string();
+ return status;
+ }
+ *fixture = segment;
+ return Status::OK();
+ }
+
+ Status prepare_scan(PreparedSegment* fixture, ReadTarget target,
VariantVersion version,
+ PreparedScan* prepared) const {
+ DORIS_CHECK(fixture != nullptr);
+ DORIS_CHECK(prepared != nullptr);
+ TabletSchemaPB schema_pb;
+ fixture->schema->to_schema_pb(&schema_pb);
+ auto query_schema = std::make_shared<TabletSchema>();
+ query_schema->init_from_pb(schema_pb);
+ query_schema->set_storage_format(fixture->schema->storage_format());
+
+ const int32_t root_id = query_schema->field_index(ROOT_UID);
+ if (root_id < 0) {
+ return Status::InternalError("Variant benchmark root column is
missing");
+ }
+ const TabletColumn& root = query_schema->column(root_id);
+ auto output_id = static_cast<ColumnId>(root_id);
+ if (target == ReadTarget::WHOLE) {
+ query_schema->mutable_column(root_id).set_variant_is_v2(version ==
VariantVersion::V2);
+ } else {
+ std::string_view relative_path;
+ switch (target) {
+ case ReadTarget::MATERIALIZED:
+ relative_path = HOT_PATH;
+ break;
+ case ReadTarget::SPARSE:
+ relative_path = SPARSE_PATH;
+ break;
+ case ReadTarget::MISSING:
+ relative_path = MISSING_PATH;
+ break;
+ case ReadTarget::WHOLE:
+ __builtin_unreachable();
+ }
+ const std::string full_path = root.name_lower_case() + "." +
std::string(relative_path);
+ DataTypePtr path_type = std::make_shared<DataTypeVariant>(
+ root.variant_max_subcolumns_count(),
root.variant_enable_doc_mode());
+ if (version == VariantVersion::V2) {
+ path_type =
std::make_shared<DataTypeVariantV2>(root.variant_max_subcolumns_count(),
+
root.variant_enable_doc_mode());
+ }
+ TabletColumn path_column = variant_util::get_column_by_type(
+ path_type, full_path,
+ variant_util::ExtraInfo {.parent_unique_id =
root.unique_id(),
+ .path_info =
PathInData(full_path)});
+ path_column.set_is_nullable(true);
+ variant_util::inherit_column_attributes(root, path_column);
+ query_schema->append_column(path_column,
TabletSchema::ColumnType::VARIANT);
+ output_id = static_cast<ColumnId>(query_schema->num_columns() - 1);
+ }
+
+ prepared->fixture = fixture;
+ prepared->query_schema = std::move(query_schema);
+ prepared->scan_schema =
std::make_shared<Schema>(prepared->query_schema->columns(),
+ std::vector<ColumnId>
{output_id});
+ prepared->output_column_id = output_id;
+ prepared->target = target;
+ return Status::OK();
+ }
+
+ Status prepare_rewrite_scan(PreparedSegment* fixture, VariantVersion
version,
+ PreparedScan* prepared) const {
+ RETURN_IF_ERROR(prepare_scan(fixture, ReadTarget::WHOLE, version,
prepared));
+ const int32_t key_id = prepared->query_schema->field_index(KEY_UID);
+ const int32_t root_id = prepared->query_schema->field_index(ROOT_UID);
+ if (key_id < 0 || root_id < 0) {
+ return Status::InternalError("Variant rewrite benchmark columns
are missing");
+ }
+ prepared->scan_schema =
+ std::make_shared<Schema>(prepared->query_schema->columns(),
+ std::vector<ColumnId>
{static_cast<ColumnId>(key_id),
+
static_cast<ColumnId>(root_id)});
+ return Status::OK();
+ }
+
+ Status rewrite_segment(const PreparedScan& prepared, VariantLayout layout,
+ VariantVersion version, const std::string&
destination_path,
+ RewriteResult* result) const {
+ DORIS_CHECK(prepared.fixture != nullptr);
+ DORIS_CHECK(result != nullptr);
+ if (segment_count() != 1) {
+ return Status::InternalError(
+ "Variant rewrite benchmark requires one source segment,
got {}",
+ segment_count());
+ }
+
+ const auto scan_init_start = std::chrono::steady_clock::now();
+ StorageReadOptions read_options;
+ read_options.stats = &result->statistics;
+ read_options.tablet_schema = prepared.query_schema;
+ read_options.io_ctx.reader_type = ReaderType::READER_QUERY;
+ read_options.use_page_cache = true;
+ read_options.block_row_max = BATCH_ROWS;
+ read_options.preferred_block_size_bytes = 0;
+ RowwiseIteratorUPtr iterator;
+
RETURN_IF_ERROR(prepared.fixture->segment->new_iterator(prepared.scan_schema,
read_options,
+ &iterator));
+ result->scan_init_ns += elapsed_ns(scan_init_start);
+
+ const TabletSchemaSPtr& destination_schema =
_schemas[static_cast<size_t>(layout)];
+ const auto writer_init_start = std::chrono::steady_clock::now();
+ io::FileWriterPtr file_writer;
+
RETURN_IF_ERROR(io::global_local_filesystem()->create_file(destination_path,
&file_writer));
+ RowsetWriterContext rowset_context;
+ rowset_context.write_type = DataWriteType::TYPE_DIRECT;
+ rowset_context.tablet_schema = destination_schema;
+ rowset_context.tablet_path = _directory;
+
+ segment_v2::SegmentWriterOptions writer_options;
+ writer_options.num_rows_per_block = BATCH_ROWS;
+ writer_options.max_rows_per_segment = _rows_per_segment;
+ writer_options.compression_type = CompressionTypePB::LZ4;
+ writer_options.rowset_ctx = &rowset_context;
+ writer_options.write_type = DataWriteType::TYPE_DIRECT;
+ segment_v2::SegmentWriter writer(file_writer.get(), 0,
destination_schema, nullptr, nullptr,
+ writer_options, nullptr);
+ RETURN_IF_ERROR(writer.init());
+ result->writer_init_ns += elapsed_ns(writer_init_start);
+
+ const int32_t key_id = prepared.query_schema->field_index(KEY_UID);
+ const int32_t root_id = prepared.query_schema->field_index(ROOT_UID);
+ DORIS_CHECK_GE(key_id, 0);
+ DORIS_CHECK_GE(root_id, 0);
+ Block block = prepared.query_schema->create_block_by_cids(
+ {static_cast<uint32_t>(key_id),
static_cast<uint32_t>(root_id)});
+ bool checked_representation = false;
+ while (true) {
+ const auto read_start = std::chrono::steady_clock::now();
+ Status status = iterator->next_batch(&block);
+ result->read_ns += elapsed_ns(read_start);
+ if (status.is<ErrorCode::END_OF_FILE>()) {
+ break;
+ }
+ RETURN_IF_ERROR(status);
+ if (!checked_representation) {
+ const IColumn& root = *block.get_by_position(1).column;
+ const bool is_v2 = check_and_get_column<ColumnVariantV2>(root)
!= nullptr;
+ const bool is_v1 = check_and_get_column<ColumnVariant>(root)
!= nullptr;
+ if ((version == VariantVersion::V2 && !is_v2) ||
+ (version == VariantVersion::V1 && (!is_v1 || is_v2))) {
+ return Status::InternalError("Variant rewrite query
returned the wrong column");
+ }
+ checked_representation = true;
+ }
+ const uint32_t batch_rows = static_cast<uint32_t>(block.rows());
+ result->rows += batch_rows;
+ result->read_output_bytes += block.bytes();
+ const auto append_start = std::chrono::steady_clock::now();
+ RETURN_IF_ERROR(writer.append_block(&block, 0, batch_rows));
+ result->append_ns += elapsed_ns(append_start);
+ block.clear_column_data();
+ }
+ if (result->rows != _rows) {
+ return Status::InternalError("Variant rewrite read {} rows,
expected {}", result->rows,
+ _rows);
+ }
+
+ const auto finalize_start = std::chrono::steady_clock::now();
+ RETURN_IF_ERROR(writer.finalize(&result->destination_segment_bytes,
+ &result->destination_index_bytes));
+ result->finalize_ns += elapsed_ns(finalize_start);
+ result->source_segment_bytes = prepared.fixture->segment_bytes;
+ return Status::OK();
+ }
+
+ Status write_segments(VariantLayout layout, VariantVersion version,
+ const std::string& base_path, benchmark::State*
state,
+ WriteResult* result, std::vector<uint64_t>*
segment_bytes) const {
+ DORIS_CHECK(result != nullptr);
+ DORIS_CHECK(segment_bytes != nullptr);
+ const TabletSchemaSPtr& schema = _schemas[static_cast<size_t>(layout)];
+ const auto key_type = std::make_shared<DataTypeInt64>();
+ segment_bytes->clear();
+ segment_bytes->reserve(segment_count());
+
+ for (uint32_t segment_id = 0; segment_id < segment_count();
++segment_id) {
+ const uint32_t segment_begin = segment_id * _rows_per_segment;
+ const uint32_t segment_rows = rows_in_segment(segment_id);
+ const auto init_start = std::chrono::steady_clock::now();
+ {
+ io::FileWriterPtr file_writer;
+ RETURN_IF_ERROR(io::global_local_filesystem()->create_file(
+ segment_path(base_path, segment_id), &file_writer));
+ RowsetWriterContext rowset_context;
+ rowset_context.write_type = DataWriteType::TYPE_DIRECT;
+ rowset_context.tablet_schema = schema;
+ rowset_context.tablet_path = _directory;
+
+ segment_v2::SegmentWriterOptions options;
+ options.num_rows_per_block = BATCH_ROWS;
+ options.max_rows_per_segment = _rows_per_segment;
+ options.compression_type = CompressionTypePB::LZ4;
+ options.rowset_ctx = &rowset_context;
+ options.write_type = DataWriteType::TYPE_DIRECT;
+
+ segment_v2::SegmentWriter writer(file_writer.get(),
segment_id, schema, nullptr,
+ nullptr, options, nullptr);
+ RETURN_IF_ERROR(writer.init());
+ result->init_ns += elapsed_ns(init_start);
+
+ for (uint32_t offset = 0; offset < segment_rows; offset +=
BATCH_ROWS) {
+ const uint32_t batch_rows = std::min(BATCH_ROWS,
segment_rows - offset);
+ if (state != nullptr) {
+ state->PauseTiming();
+ }
+ {
+ ColumnString::MutablePtr raw_json;
+ ColumnPtr key_batch;
+ build_input_batch(segment_begin + offset, batch_rows,
&raw_json,
+ &key_batch);
+ if (state != nullptr) {
+ state->ResumeTiming();
+ }
+
+ ColumnPtr variant_column;
+ DataTypePtr variant_type;
+ const auto parse_start =
std::chrono::steady_clock::now();
+ RETURN_IF_ERROR(parse_batch(layout, version,
*raw_json, batch_rows,
+ &variant_column,
&variant_type));
+ result->parse_ns += elapsed_ns(parse_start);
+
+ Block block;
+ block.insert({std::move(key_batch), key_type, "k"});
+ block.insert({std::move(variant_column),
std::move(variant_type),
+ std::string(ROOT_NAME)});
+ const auto append_start =
std::chrono::steady_clock::now();
+ RETURN_IF_ERROR(writer.append_block(&block, 0,
batch_rows));
+ result->append_ns += elapsed_ns(append_start);
+ if (state != nullptr) {
+ state->PauseTiming();
+ }
+ }
+ if (state != nullptr) {
+ state->ResumeTiming();
+ }
+ }
+
+ uint64_t bytes = 0;
+ uint64_t index_bytes = 0;
+ const auto finalize_start = std::chrono::steady_clock::now();
+ RETURN_IF_ERROR(writer.finalize(&bytes, &index_bytes));
+ result->finalize_ns += elapsed_ns(finalize_start);
+ result->segment_bytes += bytes;
+ result->index_bytes += index_bytes;
+ if (state != nullptr) {
+ state->PauseTiming();
+ }
+ segment_bytes->push_back(bytes);
+ }
+ if (state != nullptr) {
+ state->ResumeTiming();
+ }
+ }
+ return Status::OK();
+ }
+
+ Status validate_written_segments(VariantLayout layout, VariantVersion
version,
+ const std::string& base_path,
+ const std::vector<uint64_t>&
segment_bytes,
+ LayoutCounts* counts, PreparedSegment*
retained) const {
+ DORIS_CHECK(counts != nullptr);
+ if (segment_bytes.size() != segment_count()) {
+ return Status::InternalError("Variant benchmark wrote {} segments,
expected {}",
+ segment_bytes.size(),
segment_count());
+ }
+ if (retained != nullptr && segment_count() != 1) {
+ return Status::InternalError("Only a single segment can be
retained for read cases");
+ }
+
+ *counts = LayoutCounts {};
+ uint64_t total_rows = 0;
+ for (uint32_t segment_id = 0; segment_id < segment_count();
++segment_id) {
+ PreparedSegment measured;
+ measured.schema = _schemas[static_cast<size_t>(layout)];
+ measured.path = segment_path(base_path, segment_id);
+ measured.segment_bytes = segment_bytes[segment_id];
+ const uint32_t expected_rows = rows_in_segment(segment_id);
+ RETURN_IF_ERROR(
+ open_and_validate_segment(layout, segment_id,
expected_rows, &measured));
+
+ PreparedScan scan;
+ RETURN_IF_ERROR(prepare_scan(&measured, ReadTarget::WHOLE,
version, &scan));
+ ScanResult scan_result;
+ const uint32_t global_row_offset = segment_id * _rows_per_segment;
+ RETURN_IF_ERROR(
+ scan_segment(scan, true, expected_rows, &scan_result,
true, global_row_offset));
+ RETURN_IF_ERROR(validate_route(ReadTarget::WHOLE,
scan_result.statistics));
+ total_rows += scan_result.rows;
+
+ if (segment_id == 0) {
+ *counts = measured.counts;
+ } else if (measured.counts.materialized != counts->materialized ||
+ measured.counts.sparse != counts->sparse ||
+ measured.counts.doc != counts->doc) {
+ return Status::InternalError(
+ "{} segment {} layout differs: materialized={},
sparse={}, doc={}",
+ layout_config(layout).name, segment_id,
measured.counts.materialized,
+ measured.counts.sparse, measured.counts.doc);
+ }
+
+ if (retained != nullptr) {
+ *retained = std::move(measured);
+ } else {
+ measured.segment.reset();
+
RETURN_IF_ERROR(io::global_local_filesystem()->delete_file(measured.path));
+ }
+ }
+ if (total_rows != _rows) {
+ return Status::InternalError(
+ "Variant benchmark read {} rows from {} segments, expected
{}", total_rows,
+ segment_count(), _rows);
+ }
+ return Status::OK();
+ }
+
+ Status scan_segment(const PreparedScan& prepared, bool checksum, uint32_t
expected_rows,
+ ScanResult* result, bool sample_whole = false,
+ uint32_t global_row_offset = 0) const {
+ DORIS_CHECK(prepared.fixture != nullptr);
+ DORIS_CHECK(result != nullptr);
+ DORIS_CHECK(!sample_whole || (checksum && prepared.target ==
ReadTarget::WHOLE));
+ StorageReadOptions options;
+ options.stats = &result->statistics;
+ options.tablet_schema = prepared.query_schema;
+ options.io_ctx.reader_type = ReaderType::READER_QUERY;
+ options.use_page_cache = true;
+ options.block_row_max = BATCH_ROWS;
+ options.preferred_block_size_bytes = 0;
+
+ RowwiseIteratorUPtr iterator;
+ RETURN_IF_ERROR(
+ prepared.fixture->segment->new_iterator(prepared.scan_schema,
options, &iterator));
+ Block block = prepared.query_schema->create_block_by_cids(
+ {static_cast<uint32_t>(prepared.output_column_id)});
+ while (true) {
+ Status status = iterator->next_batch(&block);
+ if (status.is<ErrorCode::END_OF_FILE>()) {
+ break;
+ }
+ RETURN_IF_ERROR(status);
+ const auto& output = block.get_by_position(0);
+ const uint32_t block_begin = result->rows;
+ result->rows += static_cast<uint32_t>(block.rows());
+ result->output_bytes += block.bytes();
+ if (checksum && prepared.target != ReadTarget::WHOLE) {
+ for (size_t row = 0; row < block.rows(); ++row) {
+ result->hits += !output.column->is_null_at(row);
+ }
+ }
+ if (checksum) {
+ if (sample_whole) {
+ for (size_t row = 0; row < block.rows(); ++row) {
+ const uint32_t local_row = block_begin +
static_cast<uint32_t>(row);
+ if (local_row % WHOLE_VALIDATION_STRIDE != 0 &&
+ local_row + 1 != expected_rows) {
+ continue;
+ }
+
+ const std::string actual =
output.type->to_string(*output.column, row);
+ const uint32_t global_row = global_row_offset +
local_row;
+ const std::string expected = make_json(global_row);
+ if (actual != expected) {
+ return Status::InternalError(
+ "Variant whole-column sample differs at
global row {}",
+ global_row);
+ }
+ }
+ } else {
+ for (size_t row = 0; row < block.rows(); ++row) {
+ result->checksum = update_checksum(
+ result->checksum,
output.type->to_string(*output.column, row));
+ }
+ }
+ } else {
+ benchmark::DoNotOptimize(result->output_bytes);
+ }
+ block.clear_column_data();
+ }
+ if (result->rows != expected_rows) {
+ return Status::InternalError("Variant benchmark read {} rows,
expected {}",
+ result->rows, expected_rows);
+ }
+ return Status::OK();
+ }
+
+private:
+ VariantSegmentBenchmarkData()
+ : _rows(configured_rows()),
+ _rows_per_segment(std::min(configured_rows_per_segment(),
_rows)),
+ _directory(benchmark_root() +
"/doris_variant_segment_benchmark_" +
+ std::to_string(getpid())) {
+ _status = initialize();
+ }
+
+ ~VariantSegmentBenchmarkData() {
+ for (auto& layouts : _writer_segments) {
+ for (auto& segment : layouts) {
+ segment.segment.reset();
+ }
+ }
+
WARN_IF_ERROR(io::global_local_filesystem()->delete_directory(_directory),
+ "Failed to clean Variant segment benchmark directory");
+ }
+
+ Status initialize() {
+ if (ExecEnv::GetInstance()->get_cache_manager() == nullptr) {
+
ExecEnv::GetInstance()->set_cache_manager(CacheManager::create_global_instance());
+ }
+ if (ExecEnv::GetInstance()->get_storage_page_cache() == nullptr) {
+ constexpr size_t CACHE_CAPACITY = 512UL << 20;
+ ExecEnv::GetInstance()->set_storage_page_cache(
+ StoragePageCache::create_global_cache(CACHE_CAPACITY, 10,
0));
+ }
+
RETURN_IF_ERROR(io::global_local_filesystem()->delete_directory(_directory));
+
RETURN_IF_ERROR(io::global_local_filesystem()->create_directory(_directory));
+ _schemas[static_cast<size_t>(VariantLayout::SPARSE16)] =
+ make_schema(VariantLayout::SPARSE16);
+ _schemas[static_cast<size_t>(VariantLayout::DOC16)] =
make_schema(VariantLayout::DOC16);
+ _schemas[static_cast<size_t>(VariantLayout::FULL)] =
make_schema(VariantLayout::FULL);
+ build_input_oracle();
+ return Status::OK();
+ }
+
+ uint64_t expected_checksum(ReadTarget target) const {
+ return _expected_checksums[static_cast<size_t>(target)];
+ }
+
+ uint32_t rows_in_segment(uint32_t segment_id) const {
+ const uint64_t begin = static_cast<uint64_t>(segment_id) *
_rows_per_segment;
+ return static_cast<uint32_t>(
+ std::min<uint64_t>(_rows_per_segment,
static_cast<uint64_t>(_rows) - begin));
+ }
+
+ std::string segment_path(const std::string& base_path, uint32_t
segment_id) const {
+ if (segment_count() == 1) {
+ return base_path;
+ }
+ return base_path + "." + std::to_string(segment_id);
+ }
+
+public:
+ static TabletSchemaSPtr make_schema(VariantLayout layout) {
+ const LayoutConfig config = layout_config(layout);
+ TabletSchemaPB schema_pb;
+ schema_pb.set_keys_type(KeysType::DUP_KEYS);
+ schema_pb.set_num_short_key_columns(1);
+
+ auto* key = schema_pb.add_column();
+ key->set_unique_id(KEY_UID);
+ key->set_name("k");
+ key->set_type("BIGINT");
+ key->set_is_key(true);
+ key->set_is_nullable(false);
+
+ auto* variant = schema_pb.add_column();
+ variant->set_unique_id(ROOT_UID);
+ variant->set_name(std::string(ROOT_NAME));
+ variant->set_type("VARIANT");
+ variant->set_is_key(false);
+ variant->set_is_nullable(false);
+ variant->set_variant_max_subcolumns_count(config.max_subcolumns);
+ variant->set_variant_max_sparse_column_statistics_size(10'000);
+ variant->set_variant_sparse_hash_shard_count(BUCKETS);
+ variant->set_variant_enable_doc_mode(config.doc_mode);
+
variant->set_variant_doc_materialization_min_rows(std::numeric_limits<int64_t>::max());
+ variant->set_variant_doc_hash_shard_count(BUCKETS);
+
+ auto schema = std::make_shared<TabletSchema>();
+ schema->init_from_pb(schema_pb);
+
schema->set_storage_format(TabletStorageFormatPB::TABLET_STORAGE_FORMAT_V3);
+ return schema;
+ }
+
+ static std::string make_json(uint32_t row) {
+ static const std::array<std::string, COLD_PATHS> cold_keys = [] {
+ std::array<std::string, COLD_PATHS> keys;
+ for (uint32_t cold = 0; cold < COLD_PATHS; ++cold) {
+ std::string suffix = std::to_string(cold);
+ suffix.insert(0, 3 - suffix.size(), '0');
+ keys[cold] = "cold_" + suffix;
+ }
+ return keys;
+ }();
+ std::array<uint16_t, FIELDS_PER_ROW - 1> cold_ids {};
+ const bool include_sparse = row % SPARSE_HIT_PERIOD == 0;
+ const size_t cold_count = FIELDS_PER_ROW - 1 - include_sparse;
+ const uint32_t start = (static_cast<uint64_t>(row) * 37) % COLD_PATHS;
+ for (size_t index = 0; index < cold_count; ++index) {
+ cold_ids[index] = static_cast<uint16_t>((start + index) %
COLD_PATHS);
+ }
+ std::sort(cold_ids.begin(), cold_ids.begin() + cold_count);
+
+ std::string json;
+ json.reserve(640);
+ json.push_back('{');
+ bool first = true;
+ const auto append_field = [&](std::string_view key, uint64_t value) {
+ if (!first) {
+ json.push_back(',');
+ }
+ first = false;
+ json.push_back('"');
+ json.append(key);
+ json += "\":";
+ json += std::to_string(value);
+ };
+ for (size_t index = 0; index < cold_count; ++index) {
+ const uint32_t cold = cold_ids[index];
+ append_field(cold_keys[cold], static_cast<uint64_t>(row) *
COLD_PATHS + cold);
+ }
+ append_field(HOT_PATH, row);
+ if (include_sparse) {
+ append_field(SPARSE_PATH, static_cast<uint64_t>(row) * 17 + 3);
+ }
+ json.push_back('}');
+ DORIS_CHECK_EQ(std::count(json.begin(), json.end(), ':'),
FIELDS_PER_ROW);
+ return json;
+ }
+
+private:
+ static void build_input_batch(uint32_t begin, uint32_t rows,
ColumnString::MutablePtr* raw_json,
+ ColumnPtr* keys) {
+ DORIS_CHECK(raw_json != nullptr);
+ DORIS_CHECK(keys != nullptr);
+ auto json_batch = ColumnString::create();
+ json_batch->reserve(rows);
+ auto key_batch = ColumnInt64::create();
+ key_batch->reserve(rows);
+ for (uint32_t offset = 0; offset < rows; ++offset) {
+ const uint32_t row = begin + offset;
+ const std::string json = make_json(row);
+ json_batch->insert_data(json.data(), json.size());
+ key_batch->insert_value(row);
+ }
+ *raw_json = std::move(json_batch);
+ *keys = std::move(key_batch);
+ }
+
+ void build_input_oracle() {
+ _segment_expected_checksums.resize(segment_count());
+ for (auto& checksums : _segment_expected_checksums) {
+ checksums.fill(FNV_OFFSET);
+ }
+ for (uint32_t segment_id = 0; segment_id < segment_count();
++segment_id) {
+ const uint32_t segment_begin = segment_id * _rows_per_segment;
+ const uint32_t segment_rows = rows_in_segment(segment_id);
+ for (uint32_t offset = 0; offset < segment_rows; ++offset) {
+ const uint32_t row = segment_begin + offset;
+ const std::string json = make_json(row);
+ const bool include_sparse = row % SPARSE_HIT_PERIOD == 0;
+ const std::string materialized = std::to_string(row);
+ const std::string sparse =
+ include_sparse ?
std::to_string(static_cast<uint64_t>(row) * 17 + 3)
+ : "NULL";
+ _input_bytes += json.size();
+ const std::array<std::string_view, 4> values {json,
materialized, sparse, "NULL"};
+ for (size_t target = 0; target < values.size(); ++target) {
+ _expected_checksums[target] =
+ update_checksum(_expected_checksums[target],
values[target]);
+ _segment_expected_checksums[segment_id][target] =
update_checksum(
+ _segment_expected_checksums[segment_id][target],
values[target]);
+ }
+ }
+ }
+ }
+
+ Status parse_batch(VariantLayout layout, VariantVersion version, const
ColumnString& raw_json,
+ uint32_t rows, ColumnPtr* column, DataTypePtr* type)
const {
+ DORIS_CHECK(column != nullptr);
+ DORIS_CHECK(type != nullptr);
+ const LayoutConfig config = layout_config(layout);
+ if (version == VariantVersion::V1) {
+ auto values = ColumnVariant::create(config.max_subcolumns,
config.doc_mode);
+ const ParseConfig parse_config = v1_parse_config(layout);
+ JsonParser parser;
+ RETURN_IF_CATCH_EXCEPTION({
+ for (size_t row = 0; row < rows; ++row) {
+ variant_util::parse_json_to_variant(*values,
raw_json.get_data_at(row), &parser,
+ parse_config);
+ }
+ values->finalize();
+ });
+ *column = std::move(values);
+ *type = std::make_shared<DataTypeVariant>(config.max_subcolumns,
config.doc_mode);
+ } else {
+ auto values = ColumnVariantV2::create();
+ RETURN_IF_CATCH_EXCEPTION({
+ JsonStringToVariantEncoder
encoder(JsonToVariantOptions::current_config());
+ for (size_t row = 0; row < rows; ++row) {
+ encoder.add_json(raw_json.get_data_at(row));
+ }
+ VariantBatchBuilder encoded = encoder.finish_batch();
+ values->insert_encoded_batch(encoded);
+ });
+ *column = std::move(values);
+ *type = std::make_shared<DataTypeVariantV2>(config.max_subcolumns,
config.doc_mode);
+ }
+ DORIS_CHECK_EQ((*column)->size(), rows);
+ return Status::OK();
+ }
+
+ Status open_and_validate_segment(VariantLayout layout, uint32_t segment_id,
+ uint32_t expected_rows, PreparedSegment*
prepared) const {
+ DORIS_CHECK(prepared != nullptr);
+ RowsetId rowset_id;
+ rowset_id.init(10'000 + static_cast<int64_t>(layout));
+
RETURN_IF_ERROR(segment_v2::Segment::open(io::global_local_filesystem(),
prepared->path,
+ 20'000 +
static_cast<int64_t>(layout), segment_id,
+ rowset_id, prepared->schema,
+ io::FileReaderOptions {},
&prepared->segment));
+ if (prepared->segment->num_rows() != expected_rows) {
+ return Status::InternalError("{} segment has {} rows, expected {}",
+ layout_config(layout).name,
prepared->segment->num_rows(),
+ expected_rows);
+ }
+
+ std::shared_ptr<SegmentFooterPB> footer;
+ OlapReaderStatistics statistics;
+ io::IOContext io_context;
+ io_context.reader_type = ReaderType::READER_QUERY;
+ RETURN_IF_ERROR(prepared->segment->_get_segment_footer(footer,
&statistics, &io_context));
+ segment_v2::ColumnMetaAccessor accessor;
+ RETURN_IF_ERROR(accessor.init(*footer,
prepared->segment->_file_reader));
+ RETURN_IF_ERROR(accessor.traverse_metas(
+ *footer,
+ [&](const ColumnMetaPB& meta) { count_layout_columns(meta,
&prepared->counts); },
+ &statistics, &io_context));
+
+ const LayoutCounts& counts = prepared->counts;
+ switch (layout) {
+ case VariantLayout::SPARSE16:
+ if (counts.materialized != 1 || counts.sparse != BUCKETS ||
counts.doc != 0) {
+ return Status::InternalError(
+ "Sparse16 layout is materialized={}, sparse={},
doc={}",
+ counts.materialized, counts.sparse, counts.doc);
+ }
+ break;
+ case VariantLayout::DOC16:
+ if (counts.materialized != 0 || counts.sparse != 0 || counts.doc
!= BUCKETS) {
+ return Status::InternalError("Doc16 layout is materialized={},
sparse={}, doc={}",
+ counts.materialized,
counts.sparse, counts.doc);
+ }
+ break;
+ case VariantLayout::FULL:
+ if (counts.materialized != CANDIDATE_PATHS || counts.sparse !=
BUCKETS ||
+ counts.doc != 0) {
+ return Status::InternalError("Full layout is materialized={},
sparse={}, doc={}",
+ counts.materialized,
counts.sparse, counts.doc);
+ }
+ break;
+ }
+ return Status::OK();
+ }
+
+ Status validate_hit_count(ReadTarget target, uint32_t hits) const {
+ if (target == ReadTarget::WHOLE) {
+ return Status::OK();
+ }
+ const uint32_t expected = expected_hits(target);
+ if (hits != expected) {
+ return Status::InternalError("{} hit count is {}, expected {}",
target_name(target),
+ hits, expected);
+ }
+ return Status::OK();
+ }
+
+ static Status validate_route(ReadTarget target, const
OlapReaderStatistics& statistics) {
+ switch (target) {
+ case ReadTarget::WHOLE:
+ if (statistics.variant_subtree_hierarchical_iter_count <= 0) {
+ return Status::InternalError("Whole-column scan did not use
HIERARCHICAL");
+ }
+ break;
+ case ReadTarget::MATERIALIZED:
+ if (statistics.variant_subtree_leaf_iter_count <= 0) {
+ return Status::InternalError("Materialized scan did not use
LEAF");
+ }
+ break;
+ case ReadTarget::SPARSE:
+ if (statistics.variant_subtree_sparse_iter_count <= 0) {
+ return Status::InternalError("Sparse scan did not use
BINARY_EXTRACT");
+ }
+ break;
+ case ReadTarget::MISSING:
+ if (statistics.variant_subtree_default_iter_count <= 0) {
+ return Status::InternalError("Global-miss scan did not use
DEFAULT_FILL");
+ }
+ break;
+ }
+ return Status::OK();
+ }
+
+ uint32_t _rows;
+ uint32_t _rows_per_segment;
+ std::string _directory;
+ Status _status = Status::OK();
+ uint64_t _input_bytes = 0;
+ std::array<uint64_t, 4> _expected_checksums {FNV_OFFSET, FNV_OFFSET,
FNV_OFFSET, FNV_OFFSET};
+ std::vector<std::array<uint64_t, 4>> _segment_expected_checksums;
+ std::array<TabletSchemaSPtr, 3> _schemas;
+ std::array<std::array<PreparedSegment, 2>, 3> _writer_segments;
+ std::array<std::array<bool, 2>, 3> _writer_attempted {};
+ std::array<std::array<std::string, 2>, 3> _writer_errors;
+ std::array<std::array<bool, 4>, 3> _read_attempted {};
+ std::array<std::array<std::string, 4>, 3> _read_errors;
+};
+
+uint64_t next_compaction_fixture_id() {
+ static uint64_t next_id = 0;
+ return ++next_id;
+}
+
+// Measures the production cumulative-compaction entry point. Input rowsets are
+// persisted by the V2 writer, while the current compactor intentionally reads
+// and writes ColumnVariant; there is no V1-input parser in this fixture.
+class VariantCompactionBenchmarkFixture {
+public:
+ VariantCompactionBenchmarkFixture()
+ : _total_rows(configured_rows()),
+ _rows_per_rowset(_total_rows / COMPACTION_INPUT_ROWSETS),
+ _fixture_id(next_compaction_fixture_id()),
+ _directory(benchmark_root() +
"/doris_variant_compaction_benchmark_" +
+ std::to_string(getpid()) + "_" +
std::to_string(_fixture_id)),
+ _tmp_directory(_directory + "/tmp"),
+
_previous_ordered_compaction(config::enable_ordered_data_compaction),
+
_previous_compaction_checksum(config::enable_compaction_checksum),
+
_previous_vertical_compaction(config::enable_vertical_compaction),
+ _previous_vertical_variant_compaction(
+ config::enable_vertical_compact_variant_subcolumns) {}
+
+ ~VariantCompactionBenchmarkFixture() {
+ _compaction.reset();
+ _input_rowsets.clear();
+ _tablet.reset();
+ _schema.reset();
+ _data_dir.reset();
+ _engine = nullptr;
+ if (_installed_runtime) {
+ ExecEnv* env = ExecEnv::GetInstance();
+ env->set_storage_engine(std::move(_previous_storage_engine));
+ env->set_tmp_file_dir(std::move(_previous_tmp_file_dirs));
+ }
+
WARN_IF_ERROR(io::global_local_filesystem()->delete_directory(_directory),
+ "Failed to clean Variant compaction benchmark
directory");
+ config::enable_ordered_data_compaction = _previous_ordered_compaction;
+ config::enable_compaction_checksum = _previous_compaction_checksum;
+ config::enable_vertical_compaction = _previous_vertical_compaction;
+ config::enable_vertical_compact_variant_subcolumns =
_previous_vertical_variant_compaction;
+ }
+
+ uint32_t total_rows() const { return _total_rows; }
+ uint32_t rows_per_rowset() const { return _rows_per_rowset; }
+ uint32_t input_segments() const { return _input_segments; }
+ CumulativeCompaction* compaction() const { return _compaction.get(); }
+
+ Status prepare() {
+ if (config::variant_storage_parse_mode != 0) {
+ return Status::InvalidArgument(
+ "Variant cumulative compaction benchmark requires "
+ "variant_storage_parse_mode=0, actual={}",
+ config::variant_storage_parse_mode);
+ }
+ if (_total_rows % COMPACTION_INPUT_ROWSETS != 0) {
+ return Status::InvalidArgument("Variant compaction rows {} must be
divisible by {}",
+ _total_rows,
COMPACTION_INPUT_ROWSETS);
+ }
+
+ ensure_variant_compaction_runtime();
+ config::enable_ordered_data_compaction = false;
+ config::enable_compaction_checksum = false;
+ config::enable_vertical_compaction = true;
+ config::enable_vertical_compact_variant_subcolumns = true;
+
RETURN_IF_ERROR(io::global_local_filesystem()->delete_directory(_directory));
+
RETURN_IF_ERROR(io::global_local_filesystem()->create_directory(_directory));
+
RETURN_IF_ERROR(io::global_local_filesystem()->create_directory(_tmp_directory));
+
+ ExecEnv* env = ExecEnv::GetInstance();
+ _previous_storage_engine = std::move(env->_storage_engine);
+ _previous_tmp_file_dirs = std::move(env->_tmp_file_dirs);
+ _installed_runtime = true;
+
+ std::vector<StorePath> tmp_paths;
+ tmp_paths.emplace_back(_tmp_directory, 100ULL << 30);
+ auto tmp_file_dirs =
std::make_unique<segment_v2::TmpFileDirs>(tmp_paths);
+ RETURN_IF_ERROR(tmp_file_dirs->init());
+ env->set_tmp_file_dir(std::move(tmp_file_dirs));
+
+ EngineOptions engine_options;
+ auto engine = std::make_unique<StorageEngine>(engine_options);
+ _engine = engine.get();
+ _data_dir = std::make_unique<DataDir>(*_engine, _directory);
+ RETURN_IF_ERROR(_data_dir->init(true));
+ env->set_storage_engine(std::move(engine));
+
+ _schema =
VariantSegmentBenchmarkData::make_schema(VariantLayout::SPARSE16);
+ auto tablet_meta = std::make_shared<TabletMeta>(_schema);
+ const int64_t tablet_id = 100'000 + static_cast<int64_t>(_fixture_id);
+ tablet_meta->_tablet_id = tablet_id;
+ tablet_meta->set_tablet_uid(TabletUid(tablet_id, tablet_id + 1));
+ _tablet = std::make_shared<Tablet>(*_engine, tablet_meta,
_data_dir.get());
+ RETURN_IF_ERROR(_tablet->init());
+
RETURN_IF_ERROR(io::global_local_filesystem()->delete_directory(_tablet->tablet_path()));
+
RETURN_IF_ERROR(io::global_local_filesystem()->create_directory(_tablet->tablet_path()));
+
+ build_oracle();
+ _input_rowsets.reserve(COMPACTION_INPUT_ROWSETS);
+ for (uint32_t index = 0; index < COMPACTION_INPUT_ROWSETS; ++index) {
+ RowsetSharedPtr rowset;
+ RETURN_IF_ERROR(write_input_rowset(index, &rowset));
+ LayoutCounts input_layout;
+ RETURN_IF_ERROR(validate_rowset_layout(rowset, _rows_per_rowset,
&input_layout));
+ _input_segments += cast_set<uint32_t>(rowset->num_segments());
+ RETURN_IF_ERROR(record_input_inodes(rowset));
+ RETURN_IF_ERROR(_tablet->add_rowset(rowset));
+ _input_disk_bytes += rowset->total_disk_size();
+ _input_rowsets.emplace_back(std::move(rowset));
+ }
+
+ _compaction = std::make_unique<CumulativeCompaction>(*_engine,
_tablet);
+ _compaction->_input_rowsets = _input_rowsets;
+ return Status::OK();
+ }
+
+ Status validate(CompactionResult* result) {
+ DORIS_CHECK(result != nullptr);
+ if (_compaction == nullptr || _compaction->_output_rowset == nullptr) {
+ return Status::InternalError("Variant compaction did not produce
an output rowset");
+ }
+ if (_compaction->_is_ordered_data_compaction) {
+ return Status::InternalError("Variant compaction used ordered
link-file compaction");
+ }
+
+ const RowsetSharedPtr& output = _compaction->_output_rowset;
+ if (output->start_version() != 0 || output->end_version() !=
COMPACTION_INPUT_ROWSETS - 1) {
+ return Status::InternalError("Compaction output version is
[{},{}], expected [0,{}]",
+ output->start_version(),
output->end_version(),
+ COMPACTION_INPUT_ROWSETS - 1);
+ }
+ if (output->num_rows() != _total_rows) {
+ return Status::InternalError("Compaction output has {} rows,
expected {}",
+ output->num_rows(), _total_rows);
+ }
+ const int32_t root_index =
output->tablet_schema()->field_index(ROOT_UID);
+ if (root_index < 0 ||
output->tablet_schema()->column(root_index).variant_is_v2()) {
+ return Status::InternalError(
+ "ProductionCurrent compaction output unexpectedly uses
Variant V2 schema");
+ }
+ RETURN_IF_ERROR(validate_physical_rewrite(output));
+ RETURN_IF_ERROR(validate_rowset_layout(output, _total_rows,
&result->output_layout));
+ RETURN_IF_ERROR(validate_output_oracle(output));
+ result->input_json_bytes = _input_json_bytes;
+ result->input_disk_bytes = _input_disk_bytes;
+ result->output_disk_bytes = output->total_disk_size();
+ result->output_segments = cast_set<uint32_t>(output->num_segments());
+ return Status::OK();
+ }
+
+private:
+ void build_oracle() {
+ for (uint32_t row = 0; row < _total_rows; ++row) {
+ const std::string json =
VariantSegmentBenchmarkData::make_json(row);
+ const std::string key = std::to_string(row);
+ _expected_whole_checksum =
update_checksum(_expected_whole_checksum, json);
+ _expected_key_checksum = update_checksum(_expected_key_checksum,
key);
+ _expected_hot_checksum = update_checksum(_expected_hot_checksum,
key);
+ const bool has_sparse = row % SPARSE_HIT_PERIOD == 0;
+ _expected_sparse_checksum = update_checksum(
+ _expected_sparse_checksum,
+ has_sparse ? std::to_string(static_cast<uint64_t>(row) *
17 + 3) : "NULL");
+ _expected_sparse_hits += has_sparse;
+ }
+ }
+
+ Status make_v2_input_block(uint32_t rowset_index, uint32_t
first_local_row, uint32_t rows,
+ Block* block) {
+ auto keys = ColumnInt64::create();
+ auto raw_json = ColumnString::create();
+ keys->reserve(rows);
+ raw_json->reserve(rows);
+ for (uint32_t local = 0; local < rows; ++local) {
+ const uint32_t global_row =
+ (first_local_row + local) * COMPACTION_INPUT_ROWSETS +
rowset_index;
+ const std::string json =
VariantSegmentBenchmarkData::make_json(global_row);
+ keys->insert_value(global_row);
+ raw_json->insert_data(json.data(), json.size());
+ _input_json_bytes += json.size();
+ }
+
+ auto values = ColumnVariantV2::create();
+ RETURN_IF_CATCH_EXCEPTION({
+ JsonStringToVariantEncoder
encoder(JsonToVariantOptions::current_config());
+ for (uint32_t row = 0; row < rows; ++row) {
+ encoder.add_json(raw_json->get_data_at(row));
+ }
+ VariantBatchBuilder encoded = encoder.finish_batch();
+ values->insert_encoded_batch(encoded);
+ });
+ block->insert({std::move(keys), std::make_shared<DataTypeInt64>(),
"k"});
+ block->insert({std::move(values),
std::make_shared<DataTypeVariantV2>(1, false),
+ std::string(ROOT_NAME)});
+ return Status::OK();
+ }
+
+ Status write_input_rowset(uint32_t rowset_index, RowsetSharedPtr* rowset) {
+ RowsetWriterContext context;
+ RowsetId rowset_id;
+ rowset_id.init(static_cast<int64_t>(_fixture_id) * 100 + rowset_index
+ 1);
+ context.rowset_id = rowset_id;
+ context.rowset_type = BETA_ROWSET;
+ context.data_dir = _data_dir.get();
+ context.rowset_state = VISIBLE;
+ context.tablet_schema = _schema;
+ context.tablet_path = _tablet->tablet_path();
+ context.tablet_id = _tablet->tablet_id();
+ context.tablet_uid = _tablet->tablet_uid();
+ context.tablet = _tablet;
+ context.version = Version(rowset_index, rowset_index);
+ context.segments_overlap = NONOVERLAPPING;
+ context.max_rows_per_segment = _rows_per_rowset;
+ context.write_type = DataWriteType::TYPE_DIRECT;
+
+ auto writer_result = RowsetFactory::create_rowset_writer(*_engine,
context, false);
+ if (!writer_result.has_value()) {
+ return writer_result.error();
+ }
+ auto writer = std::move(writer_result).value();
+ for (uint32_t first = 0; first < _rows_per_rowset; first +=
BATCH_ROWS) {
+ Block block;
+ RETURN_IF_ERROR(make_v2_input_block(
+ rowset_index, first, std::min(BATCH_ROWS, _rows_per_rowset
- first), &block));
+ RETURN_IF_ERROR(writer->add_block(&block));
+ }
+ RETURN_IF_ERROR(writer->flush());
+ RETURN_IF_ERROR(writer->build(*rowset));
+ return Status::OK();
+ }
+
+ Status validate_rowset_layout(const RowsetSharedPtr& rowset, uint32_t
expected_rows,
+ LayoutCounts* result) const {
+ auto beta_rowset = std::static_pointer_cast<BetaRowset>(rowset);
+ std::vector<segment_v2::SegmentSharedPtr> segments;
+ RETURN_IF_ERROR(beta_rowset->load_segments(&segments));
+ if (segments.empty()) {
+ return Status::InternalError("Rowset has no segments");
+ }
+ uint64_t rows = 0;
+ bool first = true;
+ for (const auto& segment : segments) {
+ LayoutCounts counts;
+ RETURN_IF_ERROR(segment->traverse_column_meta_pbs(
+ [&](const ColumnMetaPB& meta) { count_layout_columns(meta,
&counts); }));
+ RETURN_IF_ERROR(validate_sparse16_layout(counts));
+ if (first) {
+ *result = counts;
+ first = false;
+ } else if (counts.materialized != result->materialized ||
+ counts.sparse != result->sparse || counts.doc !=
result->doc) {
+ return Status::InternalError("Compaction output segment
layouts differ");
+ }
+ rows += segment->num_rows();
+ }
+ if (rows != expected_rows) {
+ return Status::InternalError("Rowset segments contain {} rows,
expected {}", rows,
+ expected_rows);
+ }
+ return Status::OK();
+ }
+
+ Status record_input_inodes(const RowsetSharedPtr& rowset) {
+ for (uint32_t segment = 0; segment < rowset->num_segments();
++segment) {
+ struct stat file_stat {};
+ const std::string path = local_segment_path(_tablet->tablet_path(),
+
rowset->rowset_id().to_string(), segment);
+ if (::stat(path.c_str(), &file_stat) != 0) {
+ return Status::IOError("stat {} failed: {}", path,
std::strerror(errno));
+ }
+ _input_inodes.push_back(file_stat.st_ino);
+ }
+ return Status::OK();
+ }
+
+ Status validate_physical_rewrite(const RowsetSharedPtr& output) const {
+ for (uint32_t segment = 0; segment < output->num_segments();
++segment) {
+ struct stat file_stat {};
+ const std::string path = local_segment_path(_tablet->tablet_path(),
+
output->rowset_id().to_string(), segment);
+ if (::stat(path.c_str(), &file_stat) != 0) {
+ return Status::IOError("stat {} failed: {}", path,
std::strerror(errno));
+ }
+ if (std::find(_input_inodes.begin(), _input_inodes.end(),
file_stat.st_ino) !=
+ _input_inodes.end()) {
+ return Status::InternalError("Compaction output segment {}
reused an input inode",
+ segment);
+ }
+ }
+ return Status::OK();
+ }
+
+ Status validate_output_oracle(const RowsetSharedPtr& output) const {
+ TabletSchemaPB schema_pb;
+ _schema->to_schema_pb(&schema_pb);
+ auto query_schema = std::make_shared<TabletSchema>();
+ query_schema->init_from_pb(schema_pb);
+ query_schema->set_storage_format(_schema->storage_format());
+ const int32_t root_index = query_schema->field_index(ROOT_UID);
+ if (root_index < 0) {
+ return Status::InternalError("Variant root is missing from
compaction read schema");
+ }
+ query_schema->mutable_column(root_index).set_variant_is_v2(false);
+ const TabletColumn root = query_schema->column(root_index);
+ const auto append_path = [&](std::string_view path) {
+ const std::string full_path = root.name_lower_case() + "." +
std::string(path);
+ TabletColumn path_column = variant_util::get_column_by_type(
+ std::make_shared<DataTypeVariant>(1, false), full_path,
+ variant_util::ExtraInfo {.parent_unique_id =
root.unique_id(),
+ .path_info =
PathInData(full_path)});
+ path_column.set_is_nullable(true);
+ variant_util::inherit_column_attributes(root, path_column);
+ query_schema->append_column(path_column,
TabletSchema::ColumnType::VARIANT);
+ return static_cast<uint32_t>(query_schema->num_columns() - 1);
+ };
+ const uint32_t hot_id = append_path(HOT_PATH);
+ const uint32_t sparse_id = append_path(SPARSE_PATH);
+ std::vector<uint32_t> return_columns {0,
static_cast<uint32_t>(root_index), hot_id,
+ sparse_id};
+
+ RowsetReaderSharedPtr reader;
+ RETURN_IF_ERROR(output->create_reader(&reader));
+ OlapReaderStatistics statistics;
+ RowsetReaderContext context;
+ context.reader_type = ReaderType::READER_QUERY;
+ context.tablet_schema = query_schema;
+ context.need_ordered_result = true;
+ context.return_columns = &return_columns;
+ context.stats = &statistics;
+ RETURN_IF_ERROR(reader->init(&context));
+
+ uint64_t whole_checksum = FNV_OFFSET;
+ uint64_t key_checksum = FNV_OFFSET;
+ uint64_t hot_checksum = FNV_OFFSET;
+ uint64_t sparse_checksum = FNV_OFFSET;
+ uint32_t rows = 0;
+ uint32_t hot_hits = 0;
+ uint32_t sparse_hits = 0;
+ bool saw_current_compactor = false;
+ while (true) {
+ Block block = query_schema->create_block_by_cids(return_columns);
+ Status status = reader->next_batch(&block);
+ if (status.is<ErrorCode::END_OF_FILE>()) {
+ break;
+ }
+ RETURN_IF_ERROR(status);
+ const auto& key = block.get_by_position(0);
+ const auto& whole = block.get_by_position(1);
+ const auto& hot = block.get_by_position(2);
+ const auto& sparse = block.get_by_position(3);
+ if (block.rows() > 0 && !saw_current_compactor) {
+ const auto& nested =
+ assert_cast<const
ColumnNullable&>(*hot.column).get_nested_column();
+ if (check_and_get_column<ColumnVariant>(nested) == nullptr ||
+ check_and_get_column<ColumnVariantV2>(nested) != nullptr) {
+ return Status::InternalError(
+ "ProductionCurrent compaction did not return
ColumnVariant");
+ }
+ saw_current_compactor = true;
+ }
+ for (size_t row = 0; row < block.rows(); ++row) {
+ whole_checksum =
+ update_checksum(whole_checksum,
whole.type->to_string(*whole.column, row));
+ key_checksum = update_checksum(key_checksum,
key.type->to_string(*key.column, row));
+ hot_checksum = update_checksum(hot_checksum,
hot.type->to_string(*hot.column, row));
+ sparse_checksum = update_checksum(sparse_checksum,
+
sparse.type->to_string(*sparse.column, row));
+ hot_hits += !hot.column->is_null_at(row);
+ sparse_hits += !sparse.column->is_null_at(row);
+ }
+ rows += static_cast<uint32_t>(block.rows());
+ }
+ if (!saw_current_compactor || rows != _total_rows || hot_hits !=
_total_rows ||
+ sparse_hits != _expected_sparse_hits || whole_checksum !=
_expected_whole_checksum ||
+ key_checksum != _expected_key_checksum || hot_checksum !=
_expected_hot_checksum ||
+ sparse_checksum != _expected_sparse_checksum) {
+ return Status::InternalError(
+ "Compaction oracle mismatch: rows={}/{}, hot_hits={}/{},
sparse_hits={}/{}, "
+ "whole_checksum={}/{}, key_checksum={}/{},
hot_checksum={}/{}, "
+ "sparse_checksum={}/{}",
+ rows, _total_rows, hot_hits, _total_rows, sparse_hits,
_expected_sparse_hits,
+ whole_checksum, _expected_whole_checksum, key_checksum,
_expected_key_checksum,
+ hot_checksum, _expected_hot_checksum, sparse_checksum,
+ _expected_sparse_checksum);
+ }
+ return Status::OK();
+ }
+
+ uint32_t _total_rows;
+ uint32_t _rows_per_rowset;
+ uint32_t _input_segments = 0;
+ uint64_t _fixture_id;
+ std::string _directory;
+ std::string _tmp_directory;
+ bool _previous_ordered_compaction;
+ bool _previous_compaction_checksum;
+ bool _previous_vertical_compaction;
+ bool _previous_vertical_variant_compaction;
+ bool _installed_runtime = false;
+ std::unique_ptr<BaseStorageEngine> _previous_storage_engine;
+ std::unique_ptr<segment_v2::TmpFileDirs> _previous_tmp_file_dirs;
+ StorageEngine* _engine = nullptr;
+ std::unique_ptr<DataDir> _data_dir;
+ TabletSchemaSPtr _schema;
+ TabletSharedPtr _tablet;
+ std::vector<RowsetSharedPtr> _input_rowsets;
+ std::vector<ino_t> _input_inodes;
+ std::unique_ptr<CumulativeCompaction> _compaction;
+ uint64_t _input_json_bytes = 0;
+ uint64_t _input_disk_bytes = 0;
+ uint64_t _expected_whole_checksum = FNV_OFFSET;
+ uint64_t _expected_key_checksum = FNV_OFFSET;
+ uint64_t _expected_hot_checksum = FNV_OFFSET;
+ uint64_t _expected_sparse_checksum = FNV_OFFSET;
+ uint32_t _expected_sparse_hits = 0;
+};
+
+bool benchmark_status(benchmark::State& state, const Status& status) {
+ if (status.ok()) {
+ return true;
+ }
+ const std::string message = status.to_string();
+ state.SkipWithError(message);
+ return false;
+}
+
+void add_common_counters(benchmark::State& state, const
VariantSegmentBenchmarkData& data) {
+ state.counters["batch_rows"] =
+ benchmark::Counter(BATCH_ROWS,
benchmark::Counter::kIsIterationInvariant);
+ state.counters["rows_per_run"] =
+ benchmark::Counter(data.rows(),
benchmark::Counter::kIsIterationInvariant);
+ state.counters["rows_per_segment"] =
+ benchmark::Counter(data.rows_per_segment(),
benchmark::Counter::kIsIterationInvariant);
+ state.counters["segment_count"] =
+ benchmark::Counter(data.segment_count(),
benchmark::Counter::kIsIterationInvariant);
+ state.counters["input_bytes_per_row"] =
+ benchmark::Counter(static_cast<double>(data.input_bytes()) /
data.rows(),
+ benchmark::Counter::kIsIterationInvariant);
+}
+
+void BM_VariantIngestToSegment(benchmark::State& state, VariantLayout layout,
+ VariantVersion version) {
+ VariantSegmentBenchmarkData* data = nullptr;
+ std::string path;
+ WriteResult result;
+ std::vector<uint64_t> segment_bytes;
+ LayoutCounts measured_counts;
+ bool completed = false;
+ for (auto _ : state) {
+ benchmark::DoNotOptimize(_);
+ state.PauseTiming();
+ data = &VariantSegmentBenchmarkData::instance();
+ Status status = data->status();
+ if (status.ok()) {
+ status = data->ensure_writer_warmup(layout, version, nullptr);
+ }
+ if (status.ok()) {
+ path = data->measured_segment_path(layout, version, state.name());
+ status = data->delete_segment_files(path);
+ }
+ state.ResumeTiming();
+ if (!benchmark_status(state, status)) {
+ break;
+ }
+ result = WriteResult {};
+ status = data->write_segments(layout, version, path, &state, &result,
&segment_bytes);
+ if (!benchmark_status(state, status)) {
+ break;
+ }
+ benchmark::DoNotOptimize(result.segment_bytes);
+ state.PauseTiming();
+ status = data->validate_written_segments(layout, version, path,
segment_bytes,
+ &measured_counts, nullptr);
+ state.ResumeTiming();
+ if (!benchmark_status(state, status)) {
+ break;
+ }
+ completed = true;
+ }
+ if (!completed) {
+ return;
+ }
+ add_common_counters(state, *data);
+ state.counters["whole_validation_sample_rows"] = benchmark::Counter(
+ data->whole_validation_sample_rows(),
benchmark::Counter::kIsIterationInvariant);
+ state.counters["whole_validation_sample_stride"] =
+ benchmark::Counter(WHOLE_VALIDATION_STRIDE,
benchmark::Counter::kIsIterationInvariant);
+ state.counters["whole_validation_row_count_full"] =
+ benchmark::Counter(1, benchmark::Counter::kIsIterationInvariant);
+ state.counters["whole_validation_footer_layout_full"] =
+ benchmark::Counter(1, benchmark::Counter::kIsIterationInvariant);
+ state.counters["whole_validation_route_full_scan"] =
+ benchmark::Counter(1, benchmark::Counter::kIsIterationInvariant);
+ state.counters["whole_validation_canonical_sampled"] =
+ benchmark::Counter(1, benchmark::Counter::kIsIterationInvariant);
+ state.SetItemsProcessed(static_cast<int64_t>(data->rows()) *
state.iterations());
+ state.SetBytesProcessed(static_cast<int64_t>(data->input_bytes()) *
state.iterations());
+ state.counters["segment_bytes_per_row"] =
+ benchmark::Counter(static_cast<double>(result.segment_bytes) /
data->rows(),
+ benchmark::Counter::kIsIterationInvariant);
+ state.counters["index_bytes"] =
benchmark::Counter(static_cast<double>(result.index_bytes),
+
benchmark::Counter::kIsIterationInvariant);
+ state.counters["init_ms"] = result.init_ns / 1e6;
+ state.counters["parse_ms"] = result.parse_ns / 1e6;
+ state.counters["append_ms"] = result.append_ns / 1e6;
+ state.counters["finalize_ms"] = result.finalize_ns / 1e6;
+ state.counters["materialized_columns"] = measured_counts.materialized;
+ state.counters["sparse_columns"] = measured_counts.sparse;
+ state.counters["doc_columns"] = measured_counts.doc;
+ const ParseConfig parse_config = data->v1_parse_config(layout);
+ state.counters["variant_storage_parse_mode"] =
config::variant_storage_parse_mode;
+ state.counters["v1_parse_to_doc_value"] =
+ parse_config.parse_to == ParseConfig::ParseTo::OnlyDocValueColumn;
+}
+
+void BM_VariantRead(benchmark::State& state, VariantLayout layout, ReadTarget
target,
+ VariantVersion version) {
+ VariantSegmentBenchmarkData* data = nullptr;
+ PreparedSegment* fixture = nullptr;
+ PreparedScan prepared;
+ ScanResult result;
+ bool completed = false;
+ for (auto _ : state) {
+ benchmark::DoNotOptimize(_);
+ state.PauseTiming();
+ data = &VariantSegmentBenchmarkData::instance();
+ Status status = data->status();
+ if (status.ok()) {
+ status = data->ensure_read_validation(layout, target, &fixture);
+ }
+ if (status.ok()) {
+ status = data->prepare_scan(fixture, target, version, &prepared);
+ }
+ state.ResumeTiming();
+ if (!benchmark_status(state, status)) {
+ break;
+ }
+ result = ScanResult {};
+ status = data->scan_segment(prepared, false, data->rows(), &result);
+ if (!benchmark_status(state, status)) {
+ break;
+ }
+ completed = true;
+ }
+ if (!completed) {
+ return;
+ }
+ add_common_counters(state, *data);
+ state.counters["prevalidation_whole_sample_rows_per_representation"] =
benchmark::Counter(
+ target == ReadTarget::WHOLE ? data->whole_validation_sample_rows()
: 0,
+ benchmark::Counter::kIsIterationInvariant);
+ state.counters["prevalidation_whole_sample_stride"] =
+ target == ReadTarget::WHOLE ? WHOLE_VALIDATION_STRIDE : 0;
+ state.counters["prevalidation_exact_oracle_full_rows_per_representation"] =
+ benchmark::Counter(target == ReadTarget::WHOLE ? 0 : data->rows(),
+ benchmark::Counter::kIsIterationInvariant);
+ state.counters["prevalidation_representation_scans"] = 2;
+ state.counters["prevalidation_row_count_full"] = 1;
+ state.counters["prevalidation_footer_layout_full"] = 1;
+ state.counters["prevalidation_route_full_scan"] = 1;
+ state.counters["prevalidation_canonical_sampled"] = target ==
ReadTarget::WHOLE ? 1 : 0;
+ state.counters["prevalidation_checksum_full"] = target ==
ReadTarget::WHOLE ? 0 : 1;
+ state.SetItemsProcessed(static_cast<int64_t>(result.rows));
+ if (target == ReadTarget::WHOLE) {
+ state.SetBytesProcessed(static_cast<int64_t>(data->input_bytes()));
+ }
+ state.counters["output_bytes_per_row"] =
+ benchmark::Counter(static_cast<double>(result.output_bytes) /
result.rows,
+ benchmark::Counter::kIsIterationInvariant);
+ state.counters["segment_bytes_per_row"] =
+ benchmark::Counter(static_cast<double>(fixture->segment_bytes) /
data->rows(),
+ benchmark::Counter::kIsIterationInvariant);
+ const uint32_t expected_hits = data->expected_hits(target);
+ state.counters["hit_count"] = expected_hits;
+ state.counters["hit_rate"] = static_cast<double>(expected_hits) /
result.rows;
+ state.counters["variant_storage_parse_mode"] =
config::variant_storage_parse_mode;
+ state.counters["route_leaf"] =
result.statistics.variant_subtree_leaf_iter_count;
+ state.counters["route_binary_extract"] =
result.statistics.variant_subtree_sparse_iter_count;
+ state.counters["route_default_fill"] =
result.statistics.variant_subtree_default_iter_count;
+ state.counters["route_hierarchical"] =
+ result.statistics.variant_subtree_hierarchical_iter_count;
+ state.counters["materialized_columns"] = fixture->counts.materialized;
+ state.counters["sparse_columns"] = fixture->counts.sparse;
+ state.counters["doc_columns"] = fixture->counts.doc;
+}
+
+void BM_VariantScanAndRewriteSegment(benchmark::State& state, VariantLayout
layout,
+ VariantVersion version) {
+ VariantSegmentBenchmarkData* data = nullptr;
+ PreparedSegment* source = nullptr;
+ PreparedScan prepared;
+ RewriteResult result;
+ LayoutCounts destination_counts;
+ std::string destination_path;
+ bool completed = false;
+ for (auto _ : state) {
+ benchmark::DoNotOptimize(_);
+ state.PauseTiming();
+ data = &VariantSegmentBenchmarkData::instance();
+ Status status = data->status();
+ if (status.ok()) {
+ status = data->ensure_read_validation(layout, ReadTarget::WHOLE,
&source);
+ }
+ if (status.ok()) {
+ status = data->prepare_rewrite_scan(source, version, &prepared);
+ }
+ if (status.ok()) {
+ destination_path = data->measured_segment_path(layout, version,
state.name());
+ status = data->delete_segment_files(destination_path);
+ }
+ state.ResumeTiming();
+ if (!benchmark_status(state, status)) {
+ break;
+ }
+
+ result = RewriteResult {};
+ status = data->rewrite_segment(prepared, layout, version,
destination_path, &result);
+ if (!benchmark_status(state, status)) {
+ break;
+ }
+
+ state.PauseTiming();
+ std::vector<uint64_t> destination_bytes
{result.destination_segment_bytes};
+ status = data->validate_written_segments(layout, version,
destination_path,
+ destination_bytes,
&destination_counts, nullptr);
+ state.ResumeTiming();
+ if (!benchmark_status(state, status)) {
+ break;
+ }
+ completed = true;
+ }
+ if (!completed) {
+ return;
+ }
+
+ add_common_counters(state, *data);
+ state.SetItemsProcessed(static_cast<int64_t>(result.rows) *
state.iterations());
+ state.SetBytesProcessed(static_cast<int64_t>(result.source_segment_bytes)
* state.iterations());
+ state.counters["scan_init_ms"] = result.scan_init_ns / 1e6;
+ state.counters["writer_init_ms"] = result.writer_init_ns / 1e6;
+ state.counters["read_ms"] = result.read_ns / 1e6;
+ state.counters["append_ms"] = result.append_ns / 1e6;
+ state.counters["finalize_ms"] = result.finalize_ns / 1e6;
+ state.counters["stage_sum_ms"] = (result.scan_init_ns +
result.writer_init_ns + result.read_ns +
+ result.append_ns + result.finalize_ns) /
+ 1e6;
+ state.counters["read_output_bytes_per_row"] =
+ benchmark::Counter(static_cast<double>(result.read_output_bytes) /
result.rows,
+ benchmark::Counter::kIsIterationInvariant);
+ state.counters["source_segment_bytes_per_row"] =
+
benchmark::Counter(static_cast<double>(result.source_segment_bytes) /
result.rows,
+ benchmark::Counter::kIsIterationInvariant);
+ state.counters["destination_segment_bytes_per_row"] =
+
benchmark::Counter(static_cast<double>(result.destination_segment_bytes) /
result.rows,
+ benchmark::Counter::kIsIterationInvariant);
+ state.counters["destination_index_bytes"] = result.destination_index_bytes;
+ state.counters["source_writer_variant_v1"] = 1;
+ state.counters["query_output_variant_v2"] = version == VariantVersion::V2;
+ state.counters["destination_input_variant_v2"] = version ==
VariantVersion::V2;
+ state.counters["destination_validation_row_count_full"] = 1;
+ state.counters["destination_validation_footer_layout_full"] = 1;
+ state.counters["destination_validation_canonical_sampled"] = 1;
+ state.counters["destination_validation_route_full_scan"] = 1;
+ state.counters["source_prevalidated"] = 1;
+ state.counters["source_page_cache"] = 1;
+ state.counters["variant_storage_parse_mode"] =
config::variant_storage_parse_mode;
+ state.counters["route_hierarchical"] =
+ result.statistics.variant_subtree_hierarchical_iter_count;
+ state.counters["materialized_columns"] = destination_counts.materialized;
+ state.counters["sparse_columns"] = destination_counts.sparse;
+ state.counters["doc_columns"] = destination_counts.doc;
+}
+
+void BM_VariantCumulativeCompaction(benchmark::State& state) {
+ CompactionResult result;
+ uint32_t total_rows = 0;
+ uint32_t rows_per_rowset = 0;
+ uint32_t input_segments = 0;
+ bool completed = false;
+ for (auto _ : state) {
+ benchmark::DoNotOptimize(_);
+ state.PauseTiming();
+ auto fixture = std::make_unique<VariantCompactionBenchmarkFixture>();
+ Status status = fixture->prepare();
+ if (status.ok()) {
+ total_rows = fixture->total_rows();
+ rows_per_rowset = fixture->rows_per_rowset();
+ input_segments = fixture->input_segments();
+ }
+ CumulativeCompaction* compaction = status.ok() ? fixture->compaction()
: nullptr;
+ state.ResumeTiming();
+ if (!benchmark_status(state, status)) {
+ break;
+ }
+
+ status = compaction->execute_compact();
+
+ state.PauseTiming();
+ if (status.ok()) {
+ status = fixture->validate(&result);
+ }
+ const bool ok = benchmark_status(state, status);
+ fixture.reset();
+ state.ResumeTiming();
+ if (!ok) {
+ break;
+ }
+ completed = true;
+ }
+ if (!completed) {
+ return;
+ }
+
+ state.SetItemsProcessed(static_cast<int64_t>(total_rows) *
state.iterations());
+ state.SetBytesProcessed(static_cast<int64_t>(result.input_disk_bytes) *
state.iterations());
+ state.counters["rows_per_run"] =
+ benchmark::Counter(total_rows,
benchmark::Counter::kIsIterationInvariant);
+ state.counters["input_rowsets"] =
+ benchmark::Counter(COMPACTION_INPUT_ROWSETS,
benchmark::Counter::kIsIterationInvariant);
+ state.counters["input_segments"] =
+ benchmark::Counter(input_segments,
benchmark::Counter::kIsIterationInvariant);
+ state.counters["rows_per_input_rowset"] =
+ benchmark::Counter(rows_per_rowset,
benchmark::Counter::kIsIterationInvariant);
+ state.counters["merge_ways"] =
+ benchmark::Counter(COMPACTION_INPUT_ROWSETS,
benchmark::Counter::kIsIterationInvariant);
+ state.counters["input_key_ranges_overlap"] = 1;
+ state.counters["input_json_bytes_per_row"] =
+ benchmark::Counter(static_cast<double>(result.input_json_bytes) /
total_rows,
+ benchmark::Counter::kIsIterationInvariant);
+ state.counters["input_disk_bytes_per_row"] =
+ benchmark::Counter(static_cast<double>(result.input_disk_bytes) /
total_rows,
+ benchmark::Counter::kIsIterationInvariant);
+ state.counters["output_disk_bytes_per_row"] =
+ benchmark::Counter(static_cast<double>(result.output_disk_bytes) /
total_rows,
+ benchmark::Counter::kIsIterationInvariant);
+ state.counters["output_segments"] = result.output_segments;
+ state.counters["materialized_columns"] = result.output_layout.materialized;
+ state.counters["sparse_columns"] = result.output_layout.sparse;
+ state.counters["doc_columns"] = result.output_layout.doc;
+ state.counters["input_writer_variant_v2"] = 1;
+ state.counters["persisted_compaction_variant_v2"] = 0;
+ state.counters["ordered_link_fast_path"] = 0;
+ state.counters["vertical_compaction"] = 1;
+ state.counters["vertical_variant_subcolumns"] = 1;
+ state.counters["compaction_batch_size"] = config::compaction_batch_size;
+ state.counters["vertical_columns_per_group"] =
+ config::vertical_compaction_num_columns_per_group;
+ state.counters["fresh_input_fixture"] = 1;
+ state.counters["input_footers_prevalidated"] = 1;
+ state.counters["variant_storage_parse_mode"] =
config::variant_storage_parse_mode;
+}
+
+constexpr std::array<VariantVersion, 10> ABBA_ORDER {
+ VariantVersion::V1, VariantVersion::V2, VariantVersion::V2,
VariantVersion::V1,
+ VariantVersion::V2, VariantVersion::V1, VariantVersion::V1,
VariantVersion::V2,
+ VariantVersion::V1, VariantVersion::V2,
+};
+
+template <typename Register>
+void register_abba_pair(std::string_view prefix, bool reverse, Register&&
register_one) {
+ std::array<int, 2> sample {};
+ for (VariantVersion ordered : ABBA_ORDER) {
+ VariantVersion version = ordered;
+ if (reverse) {
+ version = ordered == VariantVersion::V1 ? VariantVersion::V2 :
VariantVersion::V1;
+ }
+ const auto version_index = static_cast<size_t>(version);
+ const std::string name = std::string(prefix) + "/sample" +
+ std::to_string(++sample[version_index]) + "_"
+
+ std::string(version_name(version));
+ register_one(name,
version)->Unit(benchmark::kMillisecond)->Iterations(1)->UseRealTime();
+ }
+}
+
+bool register_variant_segment_benchmarks() {
+ const std::array<VariantLayout, 3> layouts {VariantLayout::SPARSE16,
VariantLayout::DOC16,
+ VariantLayout::FULL};
+ size_t pair_index = 0;
+ for (VariantLayout layout : layouts) {
+ const std::string prefix =
+ "BM_VariantIngestToSegment/" +
std::string(layout_config(layout).name);
+ register_abba_pair(prefix, pair_index++ % 2 != 0,
+ [layout](const std::string& name, VariantVersion
version) {
+ return benchmark::RegisterBenchmark(
+ name, [layout,
version](benchmark::State& state) {
+ BM_VariantIngestToSegment(state,
layout, version);
+ });
+ });
+ }
+ for (VariantLayout layout : layouts) {
+ const std::string prefix =
+ "BM_VariantReadWholeColumn/" +
std::string(layout_config(layout).name);
+ register_abba_pair(prefix, pair_index++ % 2 != 0,
+ [layout](const std::string& name, VariantVersion
version) {
+ return benchmark::RegisterBenchmark(
+ name, [layout,
version](benchmark::State& state) {
+ BM_VariantRead(state, layout,
ReadTarget::WHOLE,
+ version);
+ });
+ });
+ }
+ const std::array<ReadTarget, 3> exact_targets {ReadTarget::MATERIALIZED,
ReadTarget::SPARSE,
+ ReadTarget::MISSING};
+ for (ReadTarget target : exact_targets) {
+ const std::string prefix =
+ "BM_VariantReadExactPath/Sparse16/" +
std::string(target_name(target));
+ register_abba_pair(prefix, pair_index++ % 2 != 0,
+ [target](const std::string& name, VariantVersion
version) {
+ return benchmark::RegisterBenchmark(
+ name, [target,
version](benchmark::State& state) {
+ BM_VariantRead(state,
VariantLayout::SPARSE16, target,
+ version);
+ });
+ });
+ }
+ for (VariantLayout layout : layouts) {
+ const std::string prefix = "BM_VariantScanAndRewriteSegment/" +
+ std::string(layout_config(layout).name) +
"/V1WrittenSource";
+ register_abba_pair(prefix, pair_index++ % 2 != 0,
+ [layout](const std::string& name, VariantVersion
version) {
+ return benchmark::RegisterBenchmark(
+ name, [layout,
version](benchmark::State& state) {
+
BM_VariantScanAndRewriteSegment(state, layout, version);
+ });
+ });
+ }
+ constexpr uint32_t COMPACTION_SAMPLES = 5;
+ for (uint32_t sample = 1; sample <= COMPACTION_SAMPLES; ++sample) {
+ const std::string name =
+ "BM_VariantCumulativeCompaction/Overlap10Way/FreshInput/"
+ "Sparse16/InputWrittenByV2/ProductionCurrent/sample" +
+ std::to_string(sample);
+ benchmark::RegisterBenchmark(name, BM_VariantCumulativeCompaction)
+ ->Unit(benchmark::kMillisecond)
+ ->Iterations(1)
+ ->UseRealTime();
+ }
+ return true;
+}
+
+inline const bool VARIANT_SEGMENT_BENCHMARKS_REGISTERED =
register_variant_segment_benchmarks();
+
+} // namespace
+} // namespace doris::variant_segment_benchmark
diff --git a/be/benchmark/run_variant_segment_benchmark.sh
b/be/benchmark/run_variant_segment_benchmark.sh
new file mode 100755
index 00000000000..2520865c8a9
--- /dev/null
+++ b/be/benchmark/run_variant_segment_benchmark.sh
@@ -0,0 +1,124 @@
+#!/usr/bin/env bash
+
+# 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.
+
+set -euo pipefail
+
+if [[ $# -lt 2 || $# -gt 3 ]]; then
+ echo "Usage: $0 <benchmark-binary> <result-dir> [benchmark-filter]" >&2
+ exit 2
+fi
+
+benchmark_binary=$1
+result_dir=$2
+benchmark_filter=${3:-'^BM_Variant(IngestToSegment|ReadWholeColumn|ReadExactPath|ScanAndRewriteSegment)/'}
+script_path=$(realpath "${BASH_SOURCE[0]}")
+repo_root=$(git -C "$(dirname "${script_path}")" rev-parse --show-toplevel)
+be_config=${repo_root}/conf/be.conf
+benchmark_root=${DORIS_VARIANT_BENCHMARK_ROOT:-/tmp}
+thirdparty_installed_path=$(realpath "${repo_root}/thirdparty/installed")
+
+if [[ ! -x "${benchmark_binary}" ]]; then
+ echo "Benchmark binary is not executable: ${benchmark_binary}" >&2
+ exit 2
+fi
+if [[ ! -f "${be_config}" ]]; then
+ echo "Benchmark config does not exist: ${be_config}" >&2
+ exit 2
+fi
+if [[ ! -d "${benchmark_root}" || ! -w "${benchmark_root}" ]]; then
+ echo "Benchmark root must be a writable directory: ${benchmark_root}" >&2
+ exit 2
+fi
+benchmark_binary=$(realpath "${benchmark_binary}")
+benchmark_root=$(realpath "${benchmark_root}")
+java_library_path=${JAVA_HOME:?JAVA_HOME must be set}/lib/server
+runtime_library_path="${java_library_path}${LD_LIBRARY_PATH:+:${LD_LIBRARY_PATH}}"
+
+mkdir -p "${result_dir}"
+
+{
+ date --iso-8601=seconds
+ echo "repo_root=${repo_root}"
+ echo "invocation_pwd=${PWD}"
+ echo "git_head=$(git -C "${repo_root}" rev-parse HEAD)"
+ git -C "${repo_root}" status --short --branch
+ echo "benchmark_binary=${benchmark_binary}"
+ sha256sum "${benchmark_binary}"
+ echo "JAVA_HOME=${JAVA_HOME}"
+ echo "runtime_LD_LIBRARY_PATH=${runtime_library_path}"
+ LD_LIBRARY_PATH="${runtime_library_path}" ldd "${benchmark_binary}"
+ echo "be_config=${be_config}"
+ sha256sum "${be_config}"
+ echo "variant_storage_parse_mode_config:"
+ grep -nE '^[[:space:]]*variant_storage_parse_mode[[:space:]]*='
"${be_config}" ||
+ echo "<compiled default>"
+ echo "thirdparty_installed=${thirdparty_installed_path}"
+ while IFS= read -r fingerprint; do
+ echo "$(basename "${fingerprint}")=$(< "${fingerprint}")"
+ done < <(find "${thirdparty_installed_path}" -maxdepth 1 -type f \
+ -name '*-build-fingerprint.txt' | sort)
+ if [[ -f "${repo_root}/be/build_Release/CMakeCache.txt" ]]; then
+ sha256sum "${repo_root}/be/build_Release/CMakeCache.txt"
+ grep -n '^CMAKE_BUILD_TYPE:'
"${repo_root}/be/build_Release/CMakeCache.txt"
+ fi
+ echo "benchmark_root=${benchmark_root}"
+ df -h "${benchmark_root}"
+ findmnt -T "${benchmark_root}" -o TARGET,SOURCE,FSTYPE,OPTIONS
+ lsblk -o NAME,TYPE,SIZE,ROTA,MODEL,MOUNTPOINTS
+ free -h
+ swapon --show
+ uname -a
+ lscpu
+ echo "load_before=$(< /proc/loadavg)"
+ echo "rows=${DORIS_VARIANT_BENCHMARK_ROWS:-1000000}"
+ echo
"rows_per_segment=${DORIS_VARIANT_BENCHMARK_ROWS_PER_SEGMENT:-1000000}"
+ echo "filter=${benchmark_filter}"
+ echo "cpu=${DORIS_BENCHMARK_CPU:-unbound}"
+} >"${result_dir}/environment.txt"
+
+git -C "${repo_root}" diff --binary HEAD -- >"${result_dir}/source.diff"
+sha256sum "${result_dir}/source.diff" >>"${result_dir}/environment.txt"
+
+command=("${benchmark_binary}"
+ "--benchmark_filter=${benchmark_filter}"
+ "--benchmark_counters_tabular=true"
+ "--benchmark_out=${result_dir}/raw.json"
+ "--benchmark_out_format=json")
+
+if [[ -n "${DORIS_BENCHMARK_CPU:-}" ]]; then
+ command=(taskset -c "${DORIS_BENCHMARK_CPU}" "${command[@]}")
+fi
+
+DORIS_HOME="${repo_root}" \
+ DORIS_VARIANT_BENCHMARK_ROOT="${benchmark_root}" \
+ LD_LIBRARY_PATH="${runtime_library_path}" \
+ /usr/bin/time -v -o "${result_dir}/resource.txt" "${command[@]}" \
+ >"${result_dir}/stdout.txt" 2>"${result_dir}/stderr.txt"
+
+echo "load_after=$(< /proc/loadavg)" >>"${result_dir}/environment.txt"
+
+if ! grep -q '"name": "BM_Variant' "${result_dir}/raw.json"; then
+ echo "No Variant benchmark cases matched; see ${result_dir}/raw.json" >&2
+ exit 1
+fi
+
+if grep -q '"error_occurred": true' "${result_dir}/raw.json"; then
+ echo "One or more benchmark cases failed; see ${result_dir}/raw.json" >&2
+ exit 1
+fi
diff --git a/be/src/exec/common/variant_util.cpp
b/be/src/exec/common/variant_util.cpp
index a208bfb7432..e1361a69978 100644
--- a/be/src/exec/common/variant_util.cpp
+++ b/be/src/exec/common/variant_util.cpp
@@ -2316,8 +2316,6 @@ Status parse_and_materialize_variant_columns(Block&
block, const std::vector<uin
{ return _parse_and_materialize_variant_columns(block,
variant_pos, configs); });
}
-namespace {
-
ParseConfig::ParseTo select_storage_variant_parse_target(const TabletColumn&
column,
const ParseConfig&
config) {
// NestedGroup consumes the parse-time subcolumn tree to build nested
storage structures, so it
@@ -2357,8 +2355,6 @@ ParseConfig::ParseTo
select_storage_variant_parse_target(const TabletColumn& col
}
}
-} // namespace
-
Status parse_and_materialize_variant_columns(Block& block, const TabletSchema&
tablet_schema,
const std::vector<uint32_t>&
column_pos) {
std::vector<uint32_t> variant_column_pos;
diff --git a/be/src/exec/common/variant_util.h
b/be/src/exec/common/variant_util.h
index f4302146972..6e994808fd9 100644
--- a/be/src/exec/common/variant_util.h
+++ b/be/src/exec/common/variant_util.h
@@ -256,6 +256,10 @@ Status parse_and_materialize_variant_columns(Block& block,
const std::vector<uin
void parse_json_to_variant(IColumn& column, const StringRef& jsons,
JsonParser* parser,
const ParseConfig& config);
+// Select the V1 parse target used by the storage write path for the given
schema column.
+ParseConfig::ParseTo select_storage_variant_parse_target(const TabletColumn&
column,
+ const ParseConfig&
config);
+
// Parse variant columns by picking variant positions from `column_pos` and
generating ParseConfig
// based on tablet schema settings (flatten nested / doc snapshot mode).
Status parse_and_materialize_variant_columns(Block& block, const TabletSchema&
tablet_schema,
diff --git a/be/test/storage/variant/variant_util_test.cpp
b/be/test/storage/variant/variant_util_test.cpp
index 3aec0701887..b75fa78d642 100644
--- a/be/test/storage/variant/variant_util_test.cpp
+++ b/be/test/storage/variant/variant_util_test.cpp
@@ -146,6 +146,39 @@ private:
int32_t _old_value;
};
+TEST(VariantUtilTest, SelectStorageVariantParseTargetPrecedence) {
+ TabletSchema plain_schema = _make_variant_schema(false, false);
+ TabletSchema doc_schema = _make_variant_schema(true, false);
+ TabletSchema nested_doc_schema = _make_variant_schema(true, true);
+ ParseConfig parse_config;
+
+ {
+ ScopedVariantStorageParseMode parse_mode(0);
+ EXPECT_EQ(select_storage_variant_parse_target(plain_schema.column(0),
parse_config),
+ ParseConfig::ParseTo::OnlyDocValueColumn);
+
+ parse_config.deprecated_enable_flatten_nested = true;
+ EXPECT_EQ(select_storage_variant_parse_target(plain_schema.column(0),
parse_config),
+ ParseConfig::ParseTo::OnlySubcolumns);
+ EXPECT_EQ(select_storage_variant_parse_target(doc_schema.column(0),
parse_config),
+ ParseConfig::ParseTo::OnlyDocValueColumn);
+
EXPECT_EQ(select_storage_variant_parse_target(nested_doc_schema.column(0),
parse_config),
+ ParseConfig::ParseTo::OnlySubcolumns);
+ }
+
+ parse_config.deprecated_enable_flatten_nested = false;
+ {
+ ScopedVariantStorageParseMode parse_mode(1);
+ EXPECT_EQ(select_storage_variant_parse_target(plain_schema.column(0),
parse_config),
+ ParseConfig::ParseTo::OnlySubcolumns);
+ }
+ {
+ ScopedVariantStorageParseMode parse_mode(2);
+ EXPECT_EQ(select_storage_variant_parse_target(plain_schema.column(0),
parse_config),
+ ParseConfig::ParseTo::OnlyDocValueColumn);
+ }
+}
+
TEST(VariantUtilTest,
NumericConflictResolvedAsJsonbPreservesOriginalNumberText) {
// Values distributed to different tablets can reach the reader as a JSONB
conflict. JSONB
// preserves the original number text instead of formatting the value
through DOUBLE.
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
