github-actions[bot] commented on code in PR #65921: URL: https://github.com/apache/doris/pull/65921#discussion_r3631384118
########## be/benchmark/parquet/benchmark_parquet_decoder.hpp: ########## @@ -0,0 +1,479 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#pragma once + +#include <benchmark/benchmark.h> +#include <parquet/encoding.h> +#include <parquet/schema.h> + +#include <algorithm> +#include <array> +#include <cstdint> +#include <cstring> +#include <memory> +#include <numeric> +#include <stdexcept> +#include <string> +#include <type_traits> +#include <vector> + +#include "core/custom_allocator.h" +#include "format_v2/parquet/reader/native/decoder.h" +#include "parquet_benchmark_scenarios.h" +#include "util/faststring.h" +#include "util/rle_encoding.h" +#include "util/slice.h" + +namespace doris::parquet_benchmark { +namespace detail { + +constexpr size_t DECODER_ROWS = 1UL << 16; +constexpr size_t DICTIONARY_ENTRIES = 256; +constexpr size_t FIXED_BINARY_WIDTH = 16; + +struct EncodedPage { + std::vector<uint8_t> data; + std::vector<uint8_t> dictionary; + size_t dictionary_entries = 0; + size_t value_width = 0; + bool binary = false; +}; + +inline tparquet::Type::type physical_type(ValueType value_type) { + switch (value_type) { + case ValueType::INT32: + return tparquet::Type::INT32; + case ValueType::INT64: + return tparquet::Type::INT64; + case ValueType::FLOAT: + return tparquet::Type::FLOAT; + case ValueType::DOUBLE: + return tparquet::Type::DOUBLE; + case ValueType::BYTE_ARRAY: + return tparquet::Type::BYTE_ARRAY; + case ValueType::FIXED_LEN_BYTE_ARRAY: + return tparquet::Type::FIXED_LEN_BYTE_ARRAY; + } + throw std::logic_error("unknown Parquet benchmark value type"); +} + +inline tparquet::Encoding::type parquet_encoding(Encoding encoding) { + switch (encoding) { + case Encoding::PLAIN: + return tparquet::Encoding::PLAIN; + case Encoding::DICTIONARY: + return tparquet::Encoding::RLE_DICTIONARY; + case Encoding::BYTE_STREAM_SPLIT: + return tparquet::Encoding::BYTE_STREAM_SPLIT; + case Encoding::DELTA_BINARY_PACKED: + return tparquet::Encoding::DELTA_BINARY_PACKED; + case Encoding::DELTA_LENGTH_BYTE_ARRAY: + return tparquet::Encoding::DELTA_LENGTH_BYTE_ARRAY; + case Encoding::DELTA_BYTE_ARRAY: + return tparquet::Encoding::DELTA_BYTE_ARRAY; + } + throw std::logic_error("unknown Parquet benchmark encoding"); +} + +inline std::shared_ptr<::parquet::ColumnDescriptor> descriptor(::parquet::Type::type type, + int type_length = -1) { + auto node = type == ::parquet::Type::FIXED_LEN_BYTE_ARRAY + ? ::parquet::schema::PrimitiveNode::Make( + "value", ::parquet::Repetition::REQUIRED, type, + ::parquet::ConvertedType::NONE, type_length) + : ::parquet::schema::PrimitiveNode::Make( + "value", ::parquet::Repetition::REQUIRED, type); + return std::make_shared<::parquet::ColumnDescriptor>(node, 0, 0); +} + +template <typename T> +std::vector<T> fixed_values(size_t rows) { + std::vector<T> values(rows); + for (size_t row = 0; row < rows; ++row) { + if constexpr (std::is_floating_point_v<T>) { + values[row] = static_cast<T>((row % 1009) * 0.25 - 100.0); + } else { + values[row] = static_cast<T>((row * 17) % 1000003); + } + } + return values; +} + +inline std::vector<std::string> binary_values(size_t rows, size_t width = FIXED_BINARY_WIDTH) { + std::vector<std::string> values; + values.reserve(rows); + for (size_t row = 0; row < rows; ++row) { + std::string value(width, 'a'); + const uint64_t id = row % 1009; + memcpy(value.data(), &id, std::min(width, sizeof(id))); + values.push_back(std::move(value)); + } + return values; +} + +inline std::vector<uint8_t> encode_plain_binary(const std::vector<std::string>& values) { + size_t bytes = 0; + for (const auto& value : values) { + bytes += sizeof(uint32_t) + value.size(); + } + std::vector<uint8_t> encoded; + encoded.reserve(bytes); + for (const auto& value : values) { + const auto length = static_cast<uint32_t>(value.size()); + const auto* length_bytes = reinterpret_cast<const uint8_t*>(&length); + encoded.insert(encoded.end(), length_bytes, length_bytes + sizeof(length)); + encoded.insert(encoded.end(), value.begin(), value.end()); + } + return encoded; +} + +template <typename T> +std::vector<uint8_t> encode_plain_fixed(const std::vector<T>& values) { + std::vector<uint8_t> encoded(values.size() * sizeof(T)); + memcpy(encoded.data(), values.data(), encoded.size()); + return encoded; +} + +inline std::vector<uint8_t> encode_fixed_binary(const std::vector<std::string>& values) { + std::vector<uint8_t> encoded; + encoded.reserve(values.size() * FIXED_BINARY_WIDTH); + for (const auto& value : values) { + encoded.insert(encoded.end(), value.begin(), value.end()); + } + return encoded; +} + +inline std::vector<uint8_t> encode_byte_stream_split(const std::vector<uint8_t>& plain, + size_t value_width) { + const size_t rows = plain.size() / value_width; + std::vector<uint8_t> encoded(plain.size()); + for (size_t row = 0; row < rows; ++row) { + for (size_t byte = 0; byte < value_width; ++byte) { + encoded[byte * rows + row] = plain[row * value_width + byte]; + } + } + return encoded; +} + +template <typename ParquetType, typename T> +std::vector<uint8_t> encode_delta_fixed(const std::vector<T>& values) { + auto desc = descriptor(ParquetType::type_num); + auto encoder = ::parquet::MakeTypedEncoder<ParquetType>( + ::parquet::Encoding::DELTA_BINARY_PACKED, false, desc.get()); + encoder->Put(values.data(), static_cast<int>(values.size())); + const auto buffer = encoder->FlushValues(); + return {buffer->data(), buffer->data() + buffer->size()}; +} + +inline std::vector<uint8_t> encode_delta_binary(const std::vector<std::string>& values, + ::parquet::Encoding::type encoding) { + std::vector<::parquet::ByteArray> arrays; + arrays.reserve(values.size()); + for (const auto& value : values) { + arrays.emplace_back(static_cast<uint32_t>(value.size()), + reinterpret_cast<const uint8_t*>(value.data())); + } + auto desc = descriptor(::parquet::Type::BYTE_ARRAY); + auto encoder = + ::parquet::MakeTypedEncoder<::parquet::ByteArrayType>(encoding, false, desc.get()); + encoder->Put(arrays.data(), static_cast<int>(arrays.size())); + const auto buffer = encoder->FlushValues(); + return {buffer->data(), buffer->data() + buffer->size()}; +} + +inline std::vector<uint8_t> encode_dictionary_indices(size_t rows) { + faststring encoded_ids; + RleEncoder<uint32_t> encoder(&encoded_ids, 8); + for (size_t row = 0; row < rows; ++row) { + encoder.Put(static_cast<uint32_t>(row % DICTIONARY_ENTRIES)); + } + for (size_t padding = rows; padding % 8 != 0; ++padding) { + encoder.Put(0); + } + encoder.Flush(); + std::vector<uint8_t> result(encoded_ids.size() + 1); + result[0] = 8; + memcpy(result.data() + 1, encoded_ids.data(), encoded_ids.size()); + return result; +} + +inline EncodedPage dictionary_page(ValueType value_type) { + EncodedPage page; + page.data = encode_dictionary_indices(DECODER_ROWS); + page.dictionary_entries = DICTIONARY_ENTRIES; + page.binary = value_type == ValueType::BYTE_ARRAY; + switch (value_type) { + case ValueType::INT32: + page.value_width = sizeof(int32_t); + page.dictionary = encode_plain_fixed(fixed_values<int32_t>(DICTIONARY_ENTRIES)); + break; + case ValueType::INT64: + page.value_width = sizeof(int64_t); + page.dictionary = encode_plain_fixed(fixed_values<int64_t>(DICTIONARY_ENTRIES)); + break; + case ValueType::FLOAT: + page.value_width = sizeof(float); + page.dictionary = encode_plain_fixed(fixed_values<float>(DICTIONARY_ENTRIES)); + break; + case ValueType::DOUBLE: + page.value_width = sizeof(double); + page.dictionary = encode_plain_fixed(fixed_values<double>(DICTIONARY_ENTRIES)); + break; + case ValueType::BYTE_ARRAY: + page.value_width = FIXED_BINARY_WIDTH; + page.dictionary = encode_plain_binary(binary_values(DICTIONARY_ENTRIES)); + break; + case ValueType::FIXED_LEN_BYTE_ARRAY: + page.value_width = FIXED_BINARY_WIDTH; + page.dictionary = encode_fixed_binary(binary_values(DICTIONARY_ENTRIES)); + break; + } + return page; +} + +inline EncodedPage encoded_page(const DecoderScenario& scenario) { + if (scenario.encoding == Encoding::DICTIONARY) { + return dictionary_page(scenario.value_type); + } + + EncodedPage page; + page.binary = scenario.value_type == ValueType::BYTE_ARRAY; + switch (scenario.value_type) { + case ValueType::INT32: { + auto values = fixed_values<int32_t>(DECODER_ROWS); + page.value_width = sizeof(int32_t); + page.data = scenario.encoding == Encoding::DELTA_BINARY_PACKED + ? encode_delta_fixed<::parquet::Int32Type>(values) + : encode_plain_fixed(values); + break; + } + case ValueType::INT64: { + auto values = fixed_values<int64_t>(DECODER_ROWS); + page.value_width = sizeof(int64_t); + page.data = scenario.encoding == Encoding::DELTA_BINARY_PACKED + ? encode_delta_fixed<::parquet::Int64Type>(values) + : encode_plain_fixed(values); + break; + } + case ValueType::FLOAT: { + auto plain = encode_plain_fixed(fixed_values<float>(DECODER_ROWS)); + page.value_width = sizeof(float); + page.data = scenario.encoding == Encoding::BYTE_STREAM_SPLIT + ? encode_byte_stream_split(plain, page.value_width) + : std::move(plain); + break; + } + case ValueType::DOUBLE: { + auto plain = encode_plain_fixed(fixed_values<double>(DECODER_ROWS)); + page.value_width = sizeof(double); + page.data = scenario.encoding == Encoding::BYTE_STREAM_SPLIT + ? encode_byte_stream_split(plain, page.value_width) + : std::move(plain); + break; + } + case ValueType::BYTE_ARRAY: { + auto values = binary_values(DECODER_ROWS); + page.value_width = FIXED_BINARY_WIDTH; + if (scenario.encoding == Encoding::DELTA_LENGTH_BYTE_ARRAY) { + page.data = encode_delta_binary(values, ::parquet::Encoding::DELTA_LENGTH_BYTE_ARRAY); + } else if (scenario.encoding == Encoding::DELTA_BYTE_ARRAY) { + page.data = encode_delta_binary(values, ::parquet::Encoding::DELTA_BYTE_ARRAY); + } else { + page.data = encode_plain_binary(values); + } + break; + } + case ValueType::FIXED_LEN_BYTE_ARRAY: { + auto plain = encode_fixed_binary(binary_values(DECODER_ROWS)); + page.value_width = FIXED_BINARY_WIDTH; + page.data = scenario.encoding == Encoding::BYTE_STREAM_SPLIT + ? encode_byte_stream_split(plain, page.value_width) + : std::move(plain); + break; + } + } + return page; +} + +class FixedSink final : public ParquetFixedValueConsumer { +public: + Status consume(const uint8_t* values, size_t num_values, size_t value_width) override { + benchmark::DoNotOptimize(values); + benchmark::DoNotOptimize(num_values); + benchmark::DoNotOptimize(value_width); + consumed += num_values; + return Status::OK(); + } + + size_t consumed = 0; +}; + +class BinarySink final : public ParquetBinaryValueConsumer { +public: + Status consume(const StringRef* values, size_t num_values) override { + benchmark::DoNotOptimize(values); + benchmark::DoNotOptimize(num_values); + consumed += num_values; + return Status::OK(); + } + + Status consume_plain_byte_array( + const char* encoded_data, const uint32_t* payload_offsets, + const uint32_t* value_offsets, size_t num_values, + const std::vector<ParquetSelectionRange>& value_spans) override { + benchmark::DoNotOptimize(encoded_data); + benchmark::DoNotOptimize(payload_offsets); + benchmark::DoNotOptimize(value_offsets); + auto value_spans_data = value_spans.data(); + benchmark::DoNotOptimize(value_spans_data); + consumed += num_values; + return Status::OK(); + } + + size_t consumed = 0; +}; + +class DictionarySink final : public ParquetDictionaryValueConsumer { +public: + DictionarySink(const uint8_t* dictionary, size_t value_width) + : _dictionary(dictionary), _value_width(value_width) {} + + Status consume_indices(const uint32_t* indices, size_t num_values) override { + for (size_t row = 0; row < num_values; ++row) { + auto* value = _dictionary + indices[row] * _value_width; + benchmark::DoNotOptimize(value); + } + consumed += num_values; + return Status::OK(); + } + + Status consume_repeated(uint32_t index, size_t num_values) override { + auto* value = _dictionary + index * _value_width; + benchmark::DoNotOptimize(value); + consumed += num_values; + return Status::OK(); + } + + size_t consumed = 0; + +private: + const uint8_t* _dictionary; + const size_t _value_width; +}; + +inline ParquetSelection native_selection(const SelectionPlan& plan) { + ParquetSelection selection { + .total_values = plan.total_rows, .selected_values = plan.selected_rows, .ranges = {}}; + selection.ranges.reserve(plan.ranges.size()); + for (const auto& range : plan.ranges) { + selection.ranges.push_back({.first = range.first, .count = range.count}); + } + return selection; +} + +inline void run_decoder(benchmark::State& state, DecoderScenario scenario, int selectivity, + Pattern pattern) { + auto page = encoded_page(scenario); + const auto plan = make_selection_plan(DECODER_ROWS, selectivity, pattern); + const auto selection = native_selection(plan); + std::unique_ptr<format::parquet::native::Decoder> decoder; + auto status = format::parquet::native::Decoder::get_decoder( + physical_type(scenario.value_type), parquet_encoding(scenario.encoding), decoder); + if (!status.ok()) { + state.SkipWithError(status.to_string().c_str()); + return; + } + decoder->set_type_length(static_cast<int32_t>(page.value_width)); + decoder->set_expected_values(DECODER_ROWS); + + std::vector<uint8_t> dictionary_for_sink; + if (scenario.encoding == Encoding::DICTIONARY) { + dictionary_for_sink = page.dictionary; + auto dictionary = make_unique_buffer<uint8_t>(page.dictionary.size()); + memcpy(dictionary.get(), page.dictionary.data(), page.dictionary.size()); + status = decoder->set_dict(dictionary, static_cast<int32_t>(page.dictionary.size()), + page.dictionary_entries); + if (!status.ok()) { + state.SkipWithError(status.to_string().c_str()); + return; + } + } + + Slice encoded(page.data.data(), page.data.size()); + FixedSink fixed_sink; + BinarySink binary_sink; + DictionarySink dictionary_sink(dictionary_for_sink.data(), page.value_width); + for (auto _ : state) { + state.PauseTiming(); + status = decoder->set_data(&encoded); Review Comment: [P1] Gate timing on a correct decoder result `set_data()` is fallible for the dictionary, byte-stream-split, and DELTA decoders, but this status is overwritten by the decode call. The sinks' `consumed` fields are also never checked, and the reported work comes from `plan.selected_rows`, so a decoder can produce the wrong count or values and still publish a successful throughput sample. Check the setup status immediately and add an untimed per-case count plus deterministic value checksum before accepting measurements. ########## be/benchmark/parquet/benchmark_parquet_decoder.hpp: ########## @@ -0,0 +1,479 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#pragma once + +#include <benchmark/benchmark.h> +#include <parquet/encoding.h> +#include <parquet/schema.h> + +#include <algorithm> +#include <array> +#include <cstdint> +#include <cstring> +#include <memory> +#include <numeric> +#include <stdexcept> +#include <string> +#include <type_traits> +#include <vector> + +#include "core/custom_allocator.h" +#include "format_v2/parquet/reader/native/decoder.h" +#include "parquet_benchmark_scenarios.h" +#include "util/faststring.h" +#include "util/rle_encoding.h" +#include "util/slice.h" + +namespace doris::parquet_benchmark { +namespace detail { + +constexpr size_t DECODER_ROWS = 1UL << 16; +constexpr size_t DICTIONARY_ENTRIES = 256; +constexpr size_t FIXED_BINARY_WIDTH = 16; + +struct EncodedPage { + std::vector<uint8_t> data; + std::vector<uint8_t> dictionary; + size_t dictionary_entries = 0; + size_t value_width = 0; + bool binary = false; +}; + +inline tparquet::Type::type physical_type(ValueType value_type) { + switch (value_type) { + case ValueType::INT32: + return tparquet::Type::INT32; + case ValueType::INT64: + return tparquet::Type::INT64; + case ValueType::FLOAT: + return tparquet::Type::FLOAT; + case ValueType::DOUBLE: + return tparquet::Type::DOUBLE; + case ValueType::BYTE_ARRAY: + return tparquet::Type::BYTE_ARRAY; + case ValueType::FIXED_LEN_BYTE_ARRAY: + return tparquet::Type::FIXED_LEN_BYTE_ARRAY; + } + throw std::logic_error("unknown Parquet benchmark value type"); +} + +inline tparquet::Encoding::type parquet_encoding(Encoding encoding) { + switch (encoding) { + case Encoding::PLAIN: + return tparquet::Encoding::PLAIN; + case Encoding::DICTIONARY: + return tparquet::Encoding::RLE_DICTIONARY; + case Encoding::BYTE_STREAM_SPLIT: + return tparquet::Encoding::BYTE_STREAM_SPLIT; + case Encoding::DELTA_BINARY_PACKED: + return tparquet::Encoding::DELTA_BINARY_PACKED; + case Encoding::DELTA_LENGTH_BYTE_ARRAY: + return tparquet::Encoding::DELTA_LENGTH_BYTE_ARRAY; + case Encoding::DELTA_BYTE_ARRAY: + return tparquet::Encoding::DELTA_BYTE_ARRAY; + } + throw std::logic_error("unknown Parquet benchmark encoding"); +} + +inline std::shared_ptr<::parquet::ColumnDescriptor> descriptor(::parquet::Type::type type, + int type_length = -1) { + auto node = type == ::parquet::Type::FIXED_LEN_BYTE_ARRAY + ? ::parquet::schema::PrimitiveNode::Make( + "value", ::parquet::Repetition::REQUIRED, type, + ::parquet::ConvertedType::NONE, type_length) + : ::parquet::schema::PrimitiveNode::Make( + "value", ::parquet::Repetition::REQUIRED, type); + return std::make_shared<::parquet::ColumnDescriptor>(node, 0, 0); +} + +template <typename T> +std::vector<T> fixed_values(size_t rows) { + std::vector<T> values(rows); + for (size_t row = 0; row < rows; ++row) { + if constexpr (std::is_floating_point_v<T>) { + values[row] = static_cast<T>((row % 1009) * 0.25 - 100.0); + } else { + values[row] = static_cast<T>((row * 17) % 1000003); + } + } + return values; +} + +inline std::vector<std::string> binary_values(size_t rows, size_t width = FIXED_BINARY_WIDTH) { + std::vector<std::string> values; + values.reserve(rows); + for (size_t row = 0; row < rows; ++row) { + std::string value(width, 'a'); + const uint64_t id = row % 1009; + memcpy(value.data(), &id, std::min(width, sizeof(id))); + values.push_back(std::move(value)); + } + return values; +} + +inline std::vector<uint8_t> encode_plain_binary(const std::vector<std::string>& values) { + size_t bytes = 0; + for (const auto& value : values) { + bytes += sizeof(uint32_t) + value.size(); + } + std::vector<uint8_t> encoded; + encoded.reserve(bytes); + for (const auto& value : values) { + const auto length = static_cast<uint32_t>(value.size()); + const auto* length_bytes = reinterpret_cast<const uint8_t*>(&length); + encoded.insert(encoded.end(), length_bytes, length_bytes + sizeof(length)); + encoded.insert(encoded.end(), value.begin(), value.end()); + } + return encoded; +} + +template <typename T> +std::vector<uint8_t> encode_plain_fixed(const std::vector<T>& values) { + std::vector<uint8_t> encoded(values.size() * sizeof(T)); + memcpy(encoded.data(), values.data(), encoded.size()); + return encoded; +} + +inline std::vector<uint8_t> encode_fixed_binary(const std::vector<std::string>& values) { + std::vector<uint8_t> encoded; + encoded.reserve(values.size() * FIXED_BINARY_WIDTH); + for (const auto& value : values) { + encoded.insert(encoded.end(), value.begin(), value.end()); + } + return encoded; +} + +inline std::vector<uint8_t> encode_byte_stream_split(const std::vector<uint8_t>& plain, + size_t value_width) { + const size_t rows = plain.size() / value_width; + std::vector<uint8_t> encoded(plain.size()); + for (size_t row = 0; row < rows; ++row) { + for (size_t byte = 0; byte < value_width; ++byte) { + encoded[byte * rows + row] = plain[row * value_width + byte]; + } + } + return encoded; +} + +template <typename ParquetType, typename T> +std::vector<uint8_t> encode_delta_fixed(const std::vector<T>& values) { + auto desc = descriptor(ParquetType::type_num); + auto encoder = ::parquet::MakeTypedEncoder<ParquetType>( + ::parquet::Encoding::DELTA_BINARY_PACKED, false, desc.get()); + encoder->Put(values.data(), static_cast<int>(values.size())); + const auto buffer = encoder->FlushValues(); + return {buffer->data(), buffer->data() + buffer->size()}; +} + +inline std::vector<uint8_t> encode_delta_binary(const std::vector<std::string>& values, + ::parquet::Encoding::type encoding) { + std::vector<::parquet::ByteArray> arrays; + arrays.reserve(values.size()); + for (const auto& value : values) { + arrays.emplace_back(static_cast<uint32_t>(value.size()), + reinterpret_cast<const uint8_t*>(value.data())); + } + auto desc = descriptor(::parquet::Type::BYTE_ARRAY); + auto encoder = + ::parquet::MakeTypedEncoder<::parquet::ByteArrayType>(encoding, false, desc.get()); + encoder->Put(arrays.data(), static_cast<int>(arrays.size())); + const auto buffer = encoder->FlushValues(); + return {buffer->data(), buffer->data() + buffer->size()}; +} + +inline std::vector<uint8_t> encode_dictionary_indices(size_t rows) { + faststring encoded_ids; + RleEncoder<uint32_t> encoder(&encoded_ids, 8); + for (size_t row = 0; row < rows; ++row) { + encoder.Put(static_cast<uint32_t>(row % DICTIONARY_ENTRIES)); + } + for (size_t padding = rows; padding % 8 != 0; ++padding) { + encoder.Put(0); + } + encoder.Flush(); + std::vector<uint8_t> result(encoded_ids.size() + 1); + result[0] = 8; + memcpy(result.data() + 1, encoded_ids.data(), encoded_ids.size()); + return result; +} + +inline EncodedPage dictionary_page(ValueType value_type) { + EncodedPage page; + page.data = encode_dictionary_indices(DECODER_ROWS); + page.dictionary_entries = DICTIONARY_ENTRIES; + page.binary = value_type == ValueType::BYTE_ARRAY; + switch (value_type) { + case ValueType::INT32: + page.value_width = sizeof(int32_t); + page.dictionary = encode_plain_fixed(fixed_values<int32_t>(DICTIONARY_ENTRIES)); + break; + case ValueType::INT64: + page.value_width = sizeof(int64_t); + page.dictionary = encode_plain_fixed(fixed_values<int64_t>(DICTIONARY_ENTRIES)); + break; + case ValueType::FLOAT: + page.value_width = sizeof(float); + page.dictionary = encode_plain_fixed(fixed_values<float>(DICTIONARY_ENTRIES)); + break; + case ValueType::DOUBLE: + page.value_width = sizeof(double); + page.dictionary = encode_plain_fixed(fixed_values<double>(DICTIONARY_ENTRIES)); + break; + case ValueType::BYTE_ARRAY: + page.value_width = FIXED_BINARY_WIDTH; + page.dictionary = encode_plain_binary(binary_values(DICTIONARY_ENTRIES)); + break; + case ValueType::FIXED_LEN_BYTE_ARRAY: + page.value_width = FIXED_BINARY_WIDTH; + page.dictionary = encode_fixed_binary(binary_values(DICTIONARY_ENTRIES)); + break; + } + return page; +} + +inline EncodedPage encoded_page(const DecoderScenario& scenario) { + if (scenario.encoding == Encoding::DICTIONARY) { + return dictionary_page(scenario.value_type); + } + + EncodedPage page; + page.binary = scenario.value_type == ValueType::BYTE_ARRAY; + switch (scenario.value_type) { + case ValueType::INT32: { + auto values = fixed_values<int32_t>(DECODER_ROWS); + page.value_width = sizeof(int32_t); + page.data = scenario.encoding == Encoding::DELTA_BINARY_PACKED + ? encode_delta_fixed<::parquet::Int32Type>(values) + : encode_plain_fixed(values); + break; + } + case ValueType::INT64: { + auto values = fixed_values<int64_t>(DECODER_ROWS); + page.value_width = sizeof(int64_t); + page.data = scenario.encoding == Encoding::DELTA_BINARY_PACKED + ? encode_delta_fixed<::parquet::Int64Type>(values) + : encode_plain_fixed(values); + break; + } + case ValueType::FLOAT: { + auto plain = encode_plain_fixed(fixed_values<float>(DECODER_ROWS)); + page.value_width = sizeof(float); + page.data = scenario.encoding == Encoding::BYTE_STREAM_SPLIT + ? encode_byte_stream_split(plain, page.value_width) + : std::move(plain); + break; + } + case ValueType::DOUBLE: { + auto plain = encode_plain_fixed(fixed_values<double>(DECODER_ROWS)); + page.value_width = sizeof(double); + page.data = scenario.encoding == Encoding::BYTE_STREAM_SPLIT + ? encode_byte_stream_split(plain, page.value_width) + : std::move(plain); + break; + } + case ValueType::BYTE_ARRAY: { + auto values = binary_values(DECODER_ROWS); + page.value_width = FIXED_BINARY_WIDTH; + if (scenario.encoding == Encoding::DELTA_LENGTH_BYTE_ARRAY) { + page.data = encode_delta_binary(values, ::parquet::Encoding::DELTA_LENGTH_BYTE_ARRAY); + } else if (scenario.encoding == Encoding::DELTA_BYTE_ARRAY) { + page.data = encode_delta_binary(values, ::parquet::Encoding::DELTA_BYTE_ARRAY); + } else { + page.data = encode_plain_binary(values); + } + break; + } + case ValueType::FIXED_LEN_BYTE_ARRAY: { + auto plain = encode_fixed_binary(binary_values(DECODER_ROWS)); + page.value_width = FIXED_BINARY_WIDTH; + page.data = scenario.encoding == Encoding::BYTE_STREAM_SPLIT + ? encode_byte_stream_split(plain, page.value_width) + : std::move(plain); + break; + } + } + return page; +} + +class FixedSink final : public ParquetFixedValueConsumer { +public: + Status consume(const uint8_t* values, size_t num_values, size_t value_width) override { + benchmark::DoNotOptimize(values); + benchmark::DoNotOptimize(num_values); + benchmark::DoNotOptimize(value_width); + consumed += num_values; + return Status::OK(); + } + + size_t consumed = 0; +}; + +class BinarySink final : public ParquetBinaryValueConsumer { +public: + Status consume(const StringRef* values, size_t num_values) override { + benchmark::DoNotOptimize(values); + benchmark::DoNotOptimize(num_values); + consumed += num_values; + return Status::OK(); + } + + Status consume_plain_byte_array( + const char* encoded_data, const uint32_t* payload_offsets, + const uint32_t* value_offsets, size_t num_values, + const std::vector<ParquetSelectionRange>& value_spans) override { + benchmark::DoNotOptimize(encoded_data); + benchmark::DoNotOptimize(payload_offsets); + benchmark::DoNotOptimize(value_offsets); + auto value_spans_data = value_spans.data(); + benchmark::DoNotOptimize(value_spans_data); + consumed += num_values; + return Status::OK(); + } + + size_t consumed = 0; +}; + +class DictionarySink final : public ParquetDictionaryValueConsumer { +public: + DictionarySink(const uint8_t* dictionary, size_t value_width) + : _dictionary(dictionary), _value_width(value_width) {} + + Status consume_indices(const uint32_t* indices, size_t num_values) override { + for (size_t row = 0; row < num_values; ++row) { + auto* value = _dictionary + indices[row] * _value_width; + benchmark::DoNotOptimize(value); + } + consumed += num_values; + return Status::OK(); + } + + Status consume_repeated(uint32_t index, size_t num_values) override { + auto* value = _dictionary + index * _value_width; + benchmark::DoNotOptimize(value); + consumed += num_values; + return Status::OK(); + } + + size_t consumed = 0; + +private: + const uint8_t* _dictionary; + const size_t _value_width; +}; + +inline ParquetSelection native_selection(const SelectionPlan& plan) { + ParquetSelection selection { + .total_values = plan.total_rows, .selected_values = plan.selected_rows, .ranges = {}}; + selection.ranges.reserve(plan.ranges.size()); + for (const auto& range : plan.ranges) { + selection.ranges.push_back({.first = range.first, .count = range.count}); + } + return selection; +} + +inline void run_decoder(benchmark::State& state, DecoderScenario scenario, int selectivity, + Pattern pattern) { + auto page = encoded_page(scenario); + const auto plan = make_selection_plan(DECODER_ROWS, selectivity, pattern); + const auto selection = native_selection(plan); + std::unique_ptr<format::parquet::native::Decoder> decoder; + auto status = format::parquet::native::Decoder::get_decoder( + physical_type(scenario.value_type), parquet_encoding(scenario.encoding), decoder); + if (!status.ok()) { + state.SkipWithError(status.to_string().c_str()); + return; + } + decoder->set_type_length(static_cast<int32_t>(page.value_width)); + decoder->set_expected_values(DECODER_ROWS); + + std::vector<uint8_t> dictionary_for_sink; + if (scenario.encoding == Encoding::DICTIONARY) { + dictionary_for_sink = page.dictionary; + auto dictionary = make_unique_buffer<uint8_t>(page.dictionary.size()); + memcpy(dictionary.get(), page.dictionary.data(), page.dictionary.size()); + status = decoder->set_dict(dictionary, static_cast<int32_t>(page.dictionary.size()), + page.dictionary_entries); + if (!status.ok()) { + state.SkipWithError(status.to_string().c_str()); + return; + } + } + + Slice encoded(page.data.data(), page.data.size()); + FixedSink fixed_sink; + BinarySink binary_sink; + DictionarySink dictionary_sink(dictionary_for_sink.data(), page.value_width); + for (auto _ : state) { + state.PauseTiming(); + status = decoder->set_data(&encoded); + fixed_sink.consumed = 0; + binary_sink.consumed = 0; + dictionary_sink.consumed = 0; + state.ResumeTiming(); + if (scenario.encoding == Encoding::DICTIONARY) { + status = decoder->decode_selected_dictionary_values(selection, dictionary_sink); + } else if (page.binary) { + status = decoder->decode_selected_binary_values(selection, binary_sink); + } else { + status = decoder->decode_selected_fixed_values(selection, fixed_sink); + } + if (!status.ok()) { + state.SkipWithError(status.to_string().c_str()); + break; + } + benchmark::ClobberMemory(); + } + + state.SetItemsProcessed(static_cast<int64_t>(state.iterations()) * + static_cast<int64_t>(plan.selected_rows)); + state.SetBytesProcessed(static_cast<int64_t>(state.iterations()) * + static_cast<int64_t>(page.data.size())); + state.counters["raw_rows"] = static_cast<double>(DECODER_ROWS); + state.counters["selected_rows"] = static_cast<double>(plan.selected_rows); + state.counters["selection_ranges"] = static_cast<double>(plan.ranges.size()); + state.counters["encoded_bytes"] = static_cast<double>(page.data.size()); + state.counters["ns/raw_row"] = benchmark::Counter( Review Comment: [P1] Emit the unit named by these counters Google Benchmark defines an inverted rate as seconds per item; `->Unit(kNanosecond)` changes the benchmark time columns, not custom-counter values. Passing the row count here therefore serializes seconds/row under `ns/raw_row` and `ns/selected_row`, making both values 1e9 smaller than their documented unit. The reader has the identical calculation. Please scale the invariant input appropriately (or rename both counters) and cover the emitted JSON unit. ########## be/benchmark/parquet/benchmark_parquet_reader.hpp: ########## @@ -0,0 +1,448 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#pragma once + +#include <arrow/api.h> +#include <arrow/io/api.h> +#include <benchmark/benchmark.h> +#include <parquet/api/reader.h> +#include <parquet/arrow/writer.h> + +#include <algorithm> +#include <cstdint> +#include <filesystem> +#include <memory> +#include <mutex> +#include <set> +#include <stdexcept> +#include <string> +#include <vector> + +#include "core/assert_cast.h" +#include "core/block/block.h" +#include "core/column/column_nullable.h" +#include "core/column/column_vector.h" +#include "core/data_type/data_type_nullable.h" +#include "core/data_type/data_type_number.h" +#include "exprs/vexpr.h" +#include "exprs/vexpr_context.h" +#include "format_v2/file_reader.h" +#include "format_v2/parquet/parquet_reader.h" +#include "gen_cpp/Types_types.h" +#include "io/io_common.h" +#include "parquet_benchmark_scenarios.h" +#include "runtime/runtime_state.h" +#include "storage/index/zone_map/zonemap_eval_context.h" +#include "storage/index/zone_map/zonemap_filter_result.h" + +namespace doris::parquet_benchmark { +namespace reader_detail { + +constexpr size_t READER_ROWS = 1UL << 14; +constexpr size_t READER_ROW_GROUP_ROWS = 1UL << 12; + +inline void throw_if_error(const Status& status) { + if (!status.ok()) { + throw std::runtime_error(status.to_string()); + } +} + +inline bool is_null_row(size_t row, int null_percent, Pattern pattern) { + if (null_percent <= 0) { + return false; + } + if (pattern == Pattern::ALTERNATING) { + constexpr size_t NULL_PERIOD = 101; + const size_t null_slots = (static_cast<size_t>(null_percent) * NULL_PERIOD + 99) / 100; + return (row * 37) % NULL_PERIOD < null_slots; + } + constexpr size_t CLUSTER_ROWS = 1024; + return row % CLUSTER_ROWS < CLUSTER_ROWS * static_cast<size_t>(null_percent) / 100; +} + +inline std::shared_ptr<arrow::Array> build_int32_array(int null_percent, Pattern pattern) { + arrow::Int32Builder builder; + PARQUET_THROW_NOT_OK(builder.Reserve(READER_ROWS)); + for (size_t row = 0; row < READER_ROWS; ++row) { + if (is_null_row(row, null_percent, pattern)) { + PARQUET_THROW_NOT_OK(builder.AppendNull()); + } else { + PARQUET_THROW_NOT_OK(builder.Append(static_cast<int32_t>(row % 100))); + } + } + return builder.Finish().ValueOrDie(); +} + +inline ::parquet::Encoding::type file_encoding(Encoding encoding) { + switch (encoding) { + case Encoding::PLAIN: + return ::parquet::Encoding::PLAIN; + case Encoding::BYTE_STREAM_SPLIT: + return ::parquet::Encoding::BYTE_STREAM_SPLIT; + case Encoding::DELTA_BINARY_PACKED: + return ::parquet::Encoding::DELTA_BINARY_PACKED; + case Encoding::DICTIONARY: + return ::parquet::Encoding::RLE_DICTIONARY; + case Encoding::DELTA_LENGTH_BYTE_ARRAY: + return ::parquet::Encoding::DELTA_LENGTH_BYTE_ARRAY; + case Encoding::DELTA_BYTE_ARRAY: + return ::parquet::Encoding::DELTA_BYTE_ARRAY; + } + throw std::logic_error("unknown Parquet benchmark encoding"); +} + +inline std::string fixture_name(const ReaderScenario& scenario) { + return "v2_" + to_string(scenario.encoding) + "_null" + std::to_string(scenario.null_percent) + + "_" + to_string(scenario.null_pattern) + "_w" + std::to_string(scenario.schema_width) + + "_p" + std::to_string(scenario.predicate_position) + ".parquet"; +} + +inline void verify_fixture_encoding(const std::filesystem::path& path, + const ReaderScenario& scenario) { + auto reader = ::parquet::ParquetFileReader::OpenFile(path.string(), false); + const auto metadata = reader->metadata(); + if (metadata->num_rows() != static_cast<int64_t>(READER_ROWS) || + metadata->num_columns() != scenario.schema_width) { + throw std::runtime_error("Parquet benchmark fixture has unexpected shape: " + + path.string()); + } + const auto expected = file_encoding(scenario.encoding); + for (int row_group = 0; row_group < metadata->num_row_groups(); ++row_group) { + for (int column = 0; column < metadata->num_columns(); ++column) { + const auto encodings = metadata->RowGroup(row_group)->ColumnChunk(column)->encodings(); + if (std::ranges::find(encodings, expected) == encodings.end()) { + throw std::runtime_error("Parquet benchmark fixture did not use " + + to_string(scenario.encoding) + + " encoding: " + path.string()); + } + } + } +} + +inline std::filesystem::path ensure_fixture(const ReaderScenario& scenario) { + static std::mutex fixture_mutex; + const auto directory = + std::filesystem::temp_directory_path() / "doris_parquet_reader_benchmark"; + const auto path = directory / fixture_name(scenario); + std::lock_guard guard(fixture_mutex); + if (std::filesystem::exists(path)) { + verify_fixture_encoding(path, scenario); + return path; + } + + std::filesystem::create_directories(directory); + const auto temporary_path = path.string() + ".tmp"; + std::filesystem::remove(temporary_path); + const auto values = build_int32_array(scenario.null_percent, scenario.null_pattern); + std::vector<std::shared_ptr<arrow::Field>> fields; + std::vector<std::shared_ptr<arrow::ChunkedArray>> columns; + fields.reserve(scenario.schema_width); + columns.reserve(scenario.schema_width); + for (int column = 0; column < scenario.schema_width; ++column) { + fields.push_back(arrow::field("c" + std::to_string(column), arrow::int32(), true)); + columns.push_back(std::make_shared<arrow::ChunkedArray>(values)); + } + const auto table = arrow::Table::Make(arrow::schema(std::move(fields)), std::move(columns)); + + const auto output_result = arrow::io::FileOutputStream::Open(temporary_path); + if (!output_result.ok()) { + throw std::runtime_error(output_result.status().ToString()); + } + const auto output = *output_result; + ::parquet::WriterProperties::Builder properties; + properties.version(::parquet::ParquetVersion::PARQUET_2_6); + properties.data_page_version(::parquet::ParquetDataPageVersion::V2); + properties.compression(::parquet::Compression::UNCOMPRESSED); + properties.disable_statistics(); + if (scenario.encoding == Encoding::DICTIONARY) { + properties.enable_dictionary(); + } else { + properties.disable_dictionary(); + properties.encoding(file_encoding(scenario.encoding)); + } + PARQUET_THROW_NOT_OK(::parquet::arrow::WriteTable(*table, arrow::default_memory_pool(), output, + READER_ROW_GROUP_ROWS, properties.build())); + PARQUET_THROW_NOT_OK(output->Close()); + std::filesystem::rename(temporary_path, path); + verify_fixture_encoding(path, scenario); + return path; +} + +class Int32LessThanExpr final : public VExpr { +public: + Int32LessThanExpr(int column_id, int32_t upper_bound) + : VExpr(std::make_shared<DataTypeUInt8>(), false), + _column_id(column_id), + _upper_bound(upper_bound) {} + + Status execute_column_impl(VExprContext*, const Block* block, const Selector* selector, + size_t count, ColumnPtr& result_column) const override { + DORIS_CHECK(block != nullptr); + const auto& nullable = + assert_cast<const ColumnNullable&>(*block->get_by_position(_column_id).column); + const auto& values = assert_cast<const ColumnInt32&>(nullable.get_nested_column()); + const auto& nulls = nullable.get_null_map_data(); + auto result = ColumnUInt8::create(count, 0); + auto& matches = result->get_data(); + for (size_t row = 0; row < count; ++row) { + const size_t input_row = selector == nullptr ? row : (*selector)[row]; + matches[row] = !nulls[input_row] && values.get_element(input_row) < _upper_bound; + } + result_column = std::move(result); + return Status::OK(); + } + + bool can_execute_on_raw_fixed_values(const DataTypePtr& data_type, + int column_id) const override { + return column_id == _column_id && + remove_nullable(data_type)->get_primitive_type() == TYPE_INT; + } + + Status execute_on_raw_fixed_values(const uint8_t* values, size_t num_values, size_t value_width, + const DataTypePtr&, int column_id, + uint8_t* matches) const override { + DORIS_CHECK(column_id == _column_id); + DORIS_CHECK(value_width == sizeof(int32_t)); + const auto* typed_values = reinterpret_cast<const int32_t*>(values); Review Comment: [P1] Load raw predicate values without assuming alignment For nullable DataPage V2 input, the value slice starts after variable-length definition/repetition streams and is not guaranteed to be four-byte aligned. Reinterpreting it as `int32_t*` and indexing is undefined behavior and can fault on strict-alignment targets. Use `unaligned_load<int32_t>(values + row * sizeof(int32_t))`, matching the production raw-comparison path. ########## be/benchmark/parquet/benchmark_parquet_reader.hpp: ########## @@ -0,0 +1,448 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#pragma once + +#include <arrow/api.h> +#include <arrow/io/api.h> +#include <benchmark/benchmark.h> +#include <parquet/api/reader.h> +#include <parquet/arrow/writer.h> + +#include <algorithm> +#include <cstdint> +#include <filesystem> +#include <memory> +#include <mutex> +#include <set> +#include <stdexcept> +#include <string> +#include <vector> + +#include "core/assert_cast.h" +#include "core/block/block.h" +#include "core/column/column_nullable.h" +#include "core/column/column_vector.h" +#include "core/data_type/data_type_nullable.h" +#include "core/data_type/data_type_number.h" +#include "exprs/vexpr.h" +#include "exprs/vexpr_context.h" +#include "format_v2/file_reader.h" +#include "format_v2/parquet/parquet_reader.h" +#include "gen_cpp/Types_types.h" +#include "io/io_common.h" +#include "parquet_benchmark_scenarios.h" +#include "runtime/runtime_state.h" +#include "storage/index/zone_map/zonemap_eval_context.h" +#include "storage/index/zone_map/zonemap_filter_result.h" + +namespace doris::parquet_benchmark { +namespace reader_detail { + +constexpr size_t READER_ROWS = 1UL << 14; +constexpr size_t READER_ROW_GROUP_ROWS = 1UL << 12; + +inline void throw_if_error(const Status& status) { + if (!status.ok()) { + throw std::runtime_error(status.to_string()); + } +} + +inline bool is_null_row(size_t row, int null_percent, Pattern pattern) { + if (null_percent <= 0) { + return false; + } + if (pattern == Pattern::ALTERNATING) { + constexpr size_t NULL_PERIOD = 101; + const size_t null_slots = (static_cast<size_t>(null_percent) * NULL_PERIOD + 99) / 100; + return (row * 37) % NULL_PERIOD < null_slots; + } + constexpr size_t CLUSTER_ROWS = 1024; + return row % CLUSTER_ROWS < CLUSTER_ROWS * static_cast<size_t>(null_percent) / 100; +} + +inline std::shared_ptr<arrow::Array> build_int32_array(int null_percent, Pattern pattern) { + arrow::Int32Builder builder; + PARQUET_THROW_NOT_OK(builder.Reserve(READER_ROWS)); + for (size_t row = 0; row < READER_ROWS; ++row) { + if (is_null_row(row, null_percent, pattern)) { + PARQUET_THROW_NOT_OK(builder.AppendNull()); + } else { + PARQUET_THROW_NOT_OK(builder.Append(static_cast<int32_t>(row % 100))); + } + } + return builder.Finish().ValueOrDie(); +} + +inline ::parquet::Encoding::type file_encoding(Encoding encoding) { + switch (encoding) { + case Encoding::PLAIN: + return ::parquet::Encoding::PLAIN; + case Encoding::BYTE_STREAM_SPLIT: + return ::parquet::Encoding::BYTE_STREAM_SPLIT; + case Encoding::DELTA_BINARY_PACKED: + return ::parquet::Encoding::DELTA_BINARY_PACKED; + case Encoding::DICTIONARY: + return ::parquet::Encoding::RLE_DICTIONARY; + case Encoding::DELTA_LENGTH_BYTE_ARRAY: + return ::parquet::Encoding::DELTA_LENGTH_BYTE_ARRAY; + case Encoding::DELTA_BYTE_ARRAY: + return ::parquet::Encoding::DELTA_BYTE_ARRAY; + } + throw std::logic_error("unknown Parquet benchmark encoding"); +} + +inline std::string fixture_name(const ReaderScenario& scenario) { + return "v2_" + to_string(scenario.encoding) + "_null" + std::to_string(scenario.null_percent) + + "_" + to_string(scenario.null_pattern) + "_w" + std::to_string(scenario.schema_width) + + "_p" + std::to_string(scenario.predicate_position) + ".parquet"; +} + +inline void verify_fixture_encoding(const std::filesystem::path& path, + const ReaderScenario& scenario) { + auto reader = ::parquet::ParquetFileReader::OpenFile(path.string(), false); + const auto metadata = reader->metadata(); + if (metadata->num_rows() != static_cast<int64_t>(READER_ROWS) || + metadata->num_columns() != scenario.schema_width) { + throw std::runtime_error("Parquet benchmark fixture has unexpected shape: " + + path.string()); + } + const auto expected = file_encoding(scenario.encoding); + for (int row_group = 0; row_group < metadata->num_row_groups(); ++row_group) { + for (int column = 0; column < metadata->num_columns(); ++column) { + const auto encodings = metadata->RowGroup(row_group)->ColumnChunk(column)->encodings(); + if (std::ranges::find(encodings, expected) == encodings.end()) { Review Comment: [P1] Verify the requested data-page encoding exclusively Checking only that `encodings()` contains the requested value still accepts a mixed chunkâfor example, a dictionary column that falls back to PLAIN data pages keeps `RLE_DICTIONARY` in the list and passes this test. That silently benchmarks a different encoding under the case name. Inspect page `encoding_stats` (with the same fallback rules used by `is_fully_dictionary_encoded_chunk`) and reject/regenerate any chunk whose data pages are not all encoded as requested. ########## be/benchmark/parquet/benchmark_parquet_reader.hpp: ########## @@ -0,0 +1,448 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#pragma once + +#include <arrow/api.h> +#include <arrow/io/api.h> +#include <benchmark/benchmark.h> +#include <parquet/api/reader.h> +#include <parquet/arrow/writer.h> + +#include <algorithm> +#include <cstdint> +#include <filesystem> +#include <memory> +#include <mutex> +#include <set> +#include <stdexcept> +#include <string> +#include <vector> + +#include "core/assert_cast.h" +#include "core/block/block.h" +#include "core/column/column_nullable.h" +#include "core/column/column_vector.h" +#include "core/data_type/data_type_nullable.h" +#include "core/data_type/data_type_number.h" +#include "exprs/vexpr.h" +#include "exprs/vexpr_context.h" +#include "format_v2/file_reader.h" +#include "format_v2/parquet/parquet_reader.h" +#include "gen_cpp/Types_types.h" +#include "io/io_common.h" +#include "parquet_benchmark_scenarios.h" +#include "runtime/runtime_state.h" +#include "storage/index/zone_map/zonemap_eval_context.h" +#include "storage/index/zone_map/zonemap_filter_result.h" + +namespace doris::parquet_benchmark { +namespace reader_detail { + +constexpr size_t READER_ROWS = 1UL << 14; +constexpr size_t READER_ROW_GROUP_ROWS = 1UL << 12; + +inline void throw_if_error(const Status& status) { + if (!status.ok()) { + throw std::runtime_error(status.to_string()); + } +} + +inline bool is_null_row(size_t row, int null_percent, Pattern pattern) { + if (null_percent <= 0) { + return false; + } + if (pattern == Pattern::ALTERNATING) { + constexpr size_t NULL_PERIOD = 101; + const size_t null_slots = (static_cast<size_t>(null_percent) * NULL_PERIOD + 99) / 100; + return (row * 37) % NULL_PERIOD < null_slots; + } + constexpr size_t CLUSTER_ROWS = 1024; + return row % CLUSTER_ROWS < CLUSTER_ROWS * static_cast<size_t>(null_percent) / 100; +} + +inline std::shared_ptr<arrow::Array> build_int32_array(int null_percent, Pattern pattern) { + arrow::Int32Builder builder; + PARQUET_THROW_NOT_OK(builder.Reserve(READER_ROWS)); + for (size_t row = 0; row < READER_ROWS; ++row) { + if (is_null_row(row, null_percent, pattern)) { + PARQUET_THROW_NOT_OK(builder.AppendNull()); + } else { + PARQUET_THROW_NOT_OK(builder.Append(static_cast<int32_t>(row % 100))); + } + } + return builder.Finish().ValueOrDie(); +} + +inline ::parquet::Encoding::type file_encoding(Encoding encoding) { + switch (encoding) { + case Encoding::PLAIN: + return ::parquet::Encoding::PLAIN; + case Encoding::BYTE_STREAM_SPLIT: + return ::parquet::Encoding::BYTE_STREAM_SPLIT; + case Encoding::DELTA_BINARY_PACKED: + return ::parquet::Encoding::DELTA_BINARY_PACKED; + case Encoding::DICTIONARY: + return ::parquet::Encoding::RLE_DICTIONARY; + case Encoding::DELTA_LENGTH_BYTE_ARRAY: + return ::parquet::Encoding::DELTA_LENGTH_BYTE_ARRAY; + case Encoding::DELTA_BYTE_ARRAY: + return ::parquet::Encoding::DELTA_BYTE_ARRAY; + } + throw std::logic_error("unknown Parquet benchmark encoding"); +} + +inline std::string fixture_name(const ReaderScenario& scenario) { + return "v2_" + to_string(scenario.encoding) + "_null" + std::to_string(scenario.null_percent) + + "_" + to_string(scenario.null_pattern) + "_w" + std::to_string(scenario.schema_width) + + "_p" + std::to_string(scenario.predicate_position) + ".parquet"; +} + +inline void verify_fixture_encoding(const std::filesystem::path& path, + const ReaderScenario& scenario) { + auto reader = ::parquet::ParquetFileReader::OpenFile(path.string(), false); + const auto metadata = reader->metadata(); + if (metadata->num_rows() != static_cast<int64_t>(READER_ROWS) || + metadata->num_columns() != scenario.schema_width) { + throw std::runtime_error("Parquet benchmark fixture has unexpected shape: " + + path.string()); + } + const auto expected = file_encoding(scenario.encoding); + for (int row_group = 0; row_group < metadata->num_row_groups(); ++row_group) { + for (int column = 0; column < metadata->num_columns(); ++column) { + const auto encodings = metadata->RowGroup(row_group)->ColumnChunk(column)->encodings(); + if (std::ranges::find(encodings, expected) == encodings.end()) { + throw std::runtime_error("Parquet benchmark fixture did not use " + + to_string(scenario.encoding) + + " encoding: " + path.string()); + } + } + } +} + +inline std::filesystem::path ensure_fixture(const ReaderScenario& scenario) { + static std::mutex fixture_mutex; + const auto directory = + std::filesystem::temp_directory_path() / "doris_parquet_reader_benchmark"; + const auto path = directory / fixture_name(scenario); + std::lock_guard guard(fixture_mutex); + if (std::filesystem::exists(path)) { + verify_fixture_encoding(path, scenario); + return path; + } + + std::filesystem::create_directories(directory); + const auto temporary_path = path.string() + ".tmp"; + std::filesystem::remove(temporary_path); + const auto values = build_int32_array(scenario.null_percent, scenario.null_pattern); + std::vector<std::shared_ptr<arrow::Field>> fields; + std::vector<std::shared_ptr<arrow::ChunkedArray>> columns; + fields.reserve(scenario.schema_width); + columns.reserve(scenario.schema_width); + for (int column = 0; column < scenario.schema_width; ++column) { + fields.push_back(arrow::field("c" + std::to_string(column), arrow::int32(), true)); + columns.push_back(std::make_shared<arrow::ChunkedArray>(values)); + } + const auto table = arrow::Table::Make(arrow::schema(std::move(fields)), std::move(columns)); + + const auto output_result = arrow::io::FileOutputStream::Open(temporary_path); + if (!output_result.ok()) { + throw std::runtime_error(output_result.status().ToString()); + } + const auto output = *output_result; + ::parquet::WriterProperties::Builder properties; + properties.version(::parquet::ParquetVersion::PARQUET_2_6); + properties.data_page_version(::parquet::ParquetDataPageVersion::V2); + properties.compression(::parquet::Compression::UNCOMPRESSED); + properties.disable_statistics(); + if (scenario.encoding == Encoding::DICTIONARY) { + properties.enable_dictionary(); + } else { + properties.disable_dictionary(); + properties.encoding(file_encoding(scenario.encoding)); + } + PARQUET_THROW_NOT_OK(::parquet::arrow::WriteTable(*table, arrow::default_memory_pool(), output, + READER_ROW_GROUP_ROWS, properties.build())); + PARQUET_THROW_NOT_OK(output->Close()); + std::filesystem::rename(temporary_path, path); + verify_fixture_encoding(path, scenario); + return path; +} + +class Int32LessThanExpr final : public VExpr { +public: + Int32LessThanExpr(int column_id, int32_t upper_bound) + : VExpr(std::make_shared<DataTypeUInt8>(), false), + _column_id(column_id), + _upper_bound(upper_bound) {} + + Status execute_column_impl(VExprContext*, const Block* block, const Selector* selector, + size_t count, ColumnPtr& result_column) const override { + DORIS_CHECK(block != nullptr); + const auto& nullable = + assert_cast<const ColumnNullable&>(*block->get_by_position(_column_id).column); + const auto& values = assert_cast<const ColumnInt32&>(nullable.get_nested_column()); + const auto& nulls = nullable.get_null_map_data(); + auto result = ColumnUInt8::create(count, 0); + auto& matches = result->get_data(); + for (size_t row = 0; row < count; ++row) { + const size_t input_row = selector == nullptr ? row : (*selector)[row]; + matches[row] = !nulls[input_row] && values.get_element(input_row) < _upper_bound; + } + result_column = std::move(result); + return Status::OK(); + } + + bool can_execute_on_raw_fixed_values(const DataTypePtr& data_type, + int column_id) const override { + return column_id == _column_id && + remove_nullable(data_type)->get_primitive_type() == TYPE_INT; + } + + Status execute_on_raw_fixed_values(const uint8_t* values, size_t num_values, size_t value_width, + const DataTypePtr&, int column_id, + uint8_t* matches) const override { + DORIS_CHECK(column_id == _column_id); + DORIS_CHECK(value_width == sizeof(int32_t)); + const auto* typed_values = reinterpret_cast<const int32_t*>(values); + for (size_t row = 0; row < num_values; ++row) { + matches[row] &= typed_values[row] < _upper_bound; + } + return Status::OK(); + } + + bool can_evaluate_zonemap_filter() const override { return true; } + + ZoneMapFilterResult evaluate_zonemap_filter(const ZoneMapEvalContext& ctx) const override { + const auto zone_map = ctx.zone_map(_column_id); + if (zone_map == nullptr) { + return unsupported_zonemap_filter(ctx); + } + if (!zone_map->has_not_null) { + return ZoneMapFilterResult::kNoMatch; + } + const auto upper_bound = Field::create_field<TYPE_INT>(_upper_bound); + return zone_map->min_value >= upper_bound ? ZoneMapFilterResult::kNoMatch + : ZoneMapFilterResult::kMayMatch; + } + + bool can_evaluate_dictionary_filter() const override { return true; } + + ZoneMapFilterResult evaluate_dictionary_filter( + const DictionaryEvalContext& ctx) const override { + const auto* dictionary = ctx.slot(_column_id); + if (dictionary == nullptr) { + return ZoneMapFilterResult::kUnsupported; + } + const auto upper_bound = Field::create_field<TYPE_INT>(_upper_bound); + return std::ranges::any_of(dictionary->values, + [&](const Field& value) { return value < upper_bound; }) + ? ZoneMapFilterResult::kMayMatch + : ZoneMapFilterResult::kNoMatch; + } + + void collect_slot_column_ids(std::set<int>& column_ids) const override { + column_ids.insert(_column_id); + } + + const std::string& expr_name() const override { return _expr_name; } + +private: + const int _column_id; + const int32_t _upper_bound; + const std::string _expr_name = "ParquetBenchmarkInt32LessThan"; +}; + +inline VExprContextSPtr make_predicate(int column_position, int selectivity_percent) { + auto context = VExprContext::create_shared( + std::make_shared<Int32LessThanExpr>(column_position, selectivity_percent)); + context->_prepared = true; + context->_opened = true; + return context; +} + +inline Block make_block(const std::vector<format::ColumnDefinition>& schema) { + Block block; + for (const auto& column : schema) { + block.insert({column.type->create_column(), column.type, column.name}); + } + return block; +} + +struct ReaderSession { + RuntimeState runtime_state {TQueryOptions(), TQueryGlobals()}; + std::unique_ptr<format::parquet::ParquetReader> reader; + std::vector<format::ColumnDefinition> schema; + std::shared_ptr<format::FileScanRequest> request; +}; + +inline std::unique_ptr<ReaderSession> open_reader(const std::filesystem::path& path, + const ReaderScenario& scenario) { + auto session = std::make_unique<ReaderSession>(); + auto properties = std::make_shared<io::FileSystemProperties>(); + properties->system_type = TFileType::FILE_LOCAL; + auto description = std::make_unique<io::FileDescription>(); + description->path = path.string(); + description->file_size = static_cast<int64_t>(std::filesystem::file_size(path)); + description->range_start_offset = 0; + description->range_size = -1; + session->reader = std::make_unique<format::parquet::ParquetReader>(properties, description, + nullptr, nullptr); + throw_if_error(session->reader->init(&session->runtime_state)); + throw_if_error(session->reader->get_schema(&session->schema)); + + session->request = std::make_shared<format::FileScanRequest>(); + format::FileScanRequestBuilder request_builder(session->request.get()); + if (scenario.operation == ReaderOperation::FULL_SCAN) { + for (int column = 0; column < scenario.schema_width; ++column) { + throw_if_error(request_builder.add_non_predicate_column(format::LocalColumnId(column))); + } + } else if (scenario.operation == ReaderOperation::PREDICATE_SCAN) { + const auto predicate_id = format::LocalColumnId(scenario.predicate_position); + throw_if_error(request_builder.add_predicate_column(predicate_id)); + if (scenario.projection == Projection::PREDICATE_ONLY) { + session->request->predicate_only_columns.push_back(predicate_id); + } else { + const int payload = scenario.predicate_position == 0 ? scenario.schema_width - 1 : 0; + throw_if_error( + request_builder.add_non_predicate_column(format::LocalColumnId(payload))); + } + const auto predicate_position = session->request->local_positions.at(predicate_id).value(); + session->request->conjuncts.push_back( + make_predicate(static_cast<int>(predicate_position), scenario.selectivity_percent)); + } else { + throw_if_error(request_builder.add_non_predicate_column(format::LocalColumnId(0))); + if (scenario.schema_width > 1) { + throw_if_error(request_builder.add_non_predicate_column(format::LocalColumnId(1))); + } + } + + if (scenario.operation == ReaderOperation::LIMIT_1) { + session->reader->set_batch_size(1); + } else if (scenario.operation == ReaderOperation::LIMIT_1000) { + session->reader->set_batch_size(1000); + } + throw_if_error(session->reader->open(session->request)); + return session; +} + +inline size_t scan_reader(ReaderSession* session, const ReaderScenario& scenario) { + size_t output_rows = 0; + bool eof = false; + const size_t limit = scenario.operation == ReaderOperation::LIMIT_1 ? 1 + : scenario.operation == ReaderOperation::LIMIT_1000 ? 1000 + : 0; + while (!eof) { + auto block = make_block(session->schema); + size_t rows = 0; + throw_if_error(session->reader->get_block(&block, &rows, &eof)); + output_rows += rows; + benchmark::DoNotOptimize(block); + if (scenario.operation == ReaderOperation::OPEN_TO_FIRST_BLOCK) { + break; + } + if (limit != 0 && output_rows >= limit) { + break; + } + } + return output_rows; +} + +inline size_t raw_rows_per_iteration(const ReaderScenario& scenario) { + if (scenario.operation == ReaderOperation::LIMIT_1) { + return 1; + } + if (scenario.operation == ReaderOperation::LIMIT_1000) { + return 1000; + } + if (scenario.operation == ReaderOperation::OPEN_TO_FIRST_BLOCK) { + return format::parquet::ParquetScanScheduler::DEFAULT_READ_BATCH_SIZE; + } + return READER_ROWS; +} + +inline int projected_columns(const ReaderScenario& scenario) { + if (scenario.operation == ReaderOperation::FULL_SCAN) { + return scenario.schema_width; + } + if (scenario.operation == ReaderOperation::PREDICATE_SCAN && + scenario.projection == Projection::PREDICATE_ONLY) { + return 1; + } + return std::min(2, scenario.schema_width); +} + +inline void run_reader(benchmark::State& state, ReaderScenario scenario) { + std::filesystem::path fixture; + try { + fixture = ensure_fixture(scenario); + size_t selected_rows = 0; + for (auto _ : state) { + if (scenario.operation != ReaderOperation::OPEN_TO_FIRST_BLOCK) { + state.PauseTiming(); + } + auto session = open_reader(fixture, scenario); + if (scenario.operation != ReaderOperation::OPEN_TO_FIRST_BLOCK) { + state.ResumeTiming(); + } + selected_rows = scan_reader(session.get(), scenario); + state.PauseTiming(); + throw_if_error(session->reader->close()); + state.ResumeTiming(); Review Comment: [P2] Destroy the reader session while timing is paused `session` remains alive after `ResumeTiming()`, so destruction of the retained Parquet state/file schema, request, runtime state, and session schema is charged to every iteration. Teardown is excluded from every advertised timing boundary, and steady-state cases exclude setup as well. Reset/destroy the complete session before resuming so teardown cannot contaminate scan latency, especially for wide schemas. ########## be/benchmark/parquet/benchmark_parquet_reader.hpp: ########## @@ -0,0 +1,448 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#pragma once + +#include <arrow/api.h> +#include <arrow/io/api.h> +#include <benchmark/benchmark.h> +#include <parquet/api/reader.h> +#include <parquet/arrow/writer.h> + +#include <algorithm> +#include <cstdint> +#include <filesystem> +#include <memory> +#include <mutex> +#include <set> +#include <stdexcept> +#include <string> +#include <vector> + +#include "core/assert_cast.h" +#include "core/block/block.h" +#include "core/column/column_nullable.h" +#include "core/column/column_vector.h" +#include "core/data_type/data_type_nullable.h" +#include "core/data_type/data_type_number.h" +#include "exprs/vexpr.h" +#include "exprs/vexpr_context.h" +#include "format_v2/file_reader.h" +#include "format_v2/parquet/parquet_reader.h" +#include "gen_cpp/Types_types.h" +#include "io/io_common.h" +#include "parquet_benchmark_scenarios.h" +#include "runtime/runtime_state.h" +#include "storage/index/zone_map/zonemap_eval_context.h" +#include "storage/index/zone_map/zonemap_filter_result.h" + +namespace doris::parquet_benchmark { +namespace reader_detail { + +constexpr size_t READER_ROWS = 1UL << 14; +constexpr size_t READER_ROW_GROUP_ROWS = 1UL << 12; + +inline void throw_if_error(const Status& status) { + if (!status.ok()) { + throw std::runtime_error(status.to_string()); + } +} + +inline bool is_null_row(size_t row, int null_percent, Pattern pattern) { + if (null_percent <= 0) { + return false; + } + if (pattern == Pattern::ALTERNATING) { + constexpr size_t NULL_PERIOD = 101; + const size_t null_slots = (static_cast<size_t>(null_percent) * NULL_PERIOD + 99) / 100; Review Comment: [P2] Keep paired NULL-shape cases at the same density Ceiling the 101-slot period makes `null_1/alternating` contain 325 NULLs (1.98%) while `null_1/clustered` has 160; at `null_10` the counts are 1,785 versus 1,632 and the `< 10` output differs (1,460 versus 1,478 rows). Clustered-vs-alternating timing therefore changes both shape and amount of nullable/predicate work. Generate the exact same target NULL count for both patterns and assert it in the matrix tests. ########## be/benchmark/parquet/benchmark_parquet_reader.hpp: ########## @@ -0,0 +1,448 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#pragma once + +#include <arrow/api.h> +#include <arrow/io/api.h> +#include <benchmark/benchmark.h> +#include <parquet/api/reader.h> +#include <parquet/arrow/writer.h> + +#include <algorithm> +#include <cstdint> +#include <filesystem> +#include <memory> +#include <mutex> +#include <set> +#include <stdexcept> +#include <string> +#include <vector> + +#include "core/assert_cast.h" +#include "core/block/block.h" +#include "core/column/column_nullable.h" +#include "core/column/column_vector.h" +#include "core/data_type/data_type_nullable.h" +#include "core/data_type/data_type_number.h" +#include "exprs/vexpr.h" +#include "exprs/vexpr_context.h" +#include "format_v2/file_reader.h" +#include "format_v2/parquet/parquet_reader.h" +#include "gen_cpp/Types_types.h" +#include "io/io_common.h" +#include "parquet_benchmark_scenarios.h" +#include "runtime/runtime_state.h" +#include "storage/index/zone_map/zonemap_eval_context.h" +#include "storage/index/zone_map/zonemap_filter_result.h" + +namespace doris::parquet_benchmark { +namespace reader_detail { + +constexpr size_t READER_ROWS = 1UL << 14; +constexpr size_t READER_ROW_GROUP_ROWS = 1UL << 12; + +inline void throw_if_error(const Status& status) { + if (!status.ok()) { + throw std::runtime_error(status.to_string()); + } +} + +inline bool is_null_row(size_t row, int null_percent, Pattern pattern) { + if (null_percent <= 0) { + return false; + } + if (pattern == Pattern::ALTERNATING) { + constexpr size_t NULL_PERIOD = 101; + const size_t null_slots = (static_cast<size_t>(null_percent) * NULL_PERIOD + 99) / 100; + return (row * 37) % NULL_PERIOD < null_slots; + } + constexpr size_t CLUSTER_ROWS = 1024; + return row % CLUSTER_ROWS < CLUSTER_ROWS * static_cast<size_t>(null_percent) / 100; +} + +inline std::shared_ptr<arrow::Array> build_int32_array(int null_percent, Pattern pattern) { + arrow::Int32Builder builder; + PARQUET_THROW_NOT_OK(builder.Reserve(READER_ROWS)); + for (size_t row = 0; row < READER_ROWS; ++row) { + if (is_null_row(row, null_percent, pattern)) { + PARQUET_THROW_NOT_OK(builder.AppendNull()); + } else { + PARQUET_THROW_NOT_OK(builder.Append(static_cast<int32_t>(row % 100))); + } + } + return builder.Finish().ValueOrDie(); +} + +inline ::parquet::Encoding::type file_encoding(Encoding encoding) { + switch (encoding) { + case Encoding::PLAIN: + return ::parquet::Encoding::PLAIN; + case Encoding::BYTE_STREAM_SPLIT: + return ::parquet::Encoding::BYTE_STREAM_SPLIT; + case Encoding::DELTA_BINARY_PACKED: + return ::parquet::Encoding::DELTA_BINARY_PACKED; + case Encoding::DICTIONARY: + return ::parquet::Encoding::RLE_DICTIONARY; + case Encoding::DELTA_LENGTH_BYTE_ARRAY: + return ::parquet::Encoding::DELTA_LENGTH_BYTE_ARRAY; + case Encoding::DELTA_BYTE_ARRAY: + return ::parquet::Encoding::DELTA_BYTE_ARRAY; + } + throw std::logic_error("unknown Parquet benchmark encoding"); +} + +inline std::string fixture_name(const ReaderScenario& scenario) { + return "v2_" + to_string(scenario.encoding) + "_null" + std::to_string(scenario.null_percent) + + "_" + to_string(scenario.null_pattern) + "_w" + std::to_string(scenario.schema_width) + + "_p" + std::to_string(scenario.predicate_position) + ".parquet"; +} + +inline void verify_fixture_encoding(const std::filesystem::path& path, + const ReaderScenario& scenario) { + auto reader = ::parquet::ParquetFileReader::OpenFile(path.string(), false); + const auto metadata = reader->metadata(); + if (metadata->num_rows() != static_cast<int64_t>(READER_ROWS) || + metadata->num_columns() != scenario.schema_width) { + throw std::runtime_error("Parquet benchmark fixture has unexpected shape: " + + path.string()); + } + const auto expected = file_encoding(scenario.encoding); + for (int row_group = 0; row_group < metadata->num_row_groups(); ++row_group) { + for (int column = 0; column < metadata->num_columns(); ++column) { + const auto encodings = metadata->RowGroup(row_group)->ColumnChunk(column)->encodings(); + if (std::ranges::find(encodings, expected) == encodings.end()) { + throw std::runtime_error("Parquet benchmark fixture did not use " + + to_string(scenario.encoding) + + " encoding: " + path.string()); + } + } + } +} + +inline std::filesystem::path ensure_fixture(const ReaderScenario& scenario) { + static std::mutex fixture_mutex; + const auto directory = + std::filesystem::temp_directory_path() / "doris_parquet_reader_benchmark"; + const auto path = directory / fixture_name(scenario); + std::lock_guard guard(fixture_mutex); + if (std::filesystem::exists(path)) { + verify_fixture_encoding(path, scenario); + return path; + } + + std::filesystem::create_directories(directory); + const auto temporary_path = path.string() + ".tmp"; + std::filesystem::remove(temporary_path); + const auto values = build_int32_array(scenario.null_percent, scenario.null_pattern); + std::vector<std::shared_ptr<arrow::Field>> fields; + std::vector<std::shared_ptr<arrow::ChunkedArray>> columns; + fields.reserve(scenario.schema_width); + columns.reserve(scenario.schema_width); + for (int column = 0; column < scenario.schema_width; ++column) { + fields.push_back(arrow::field("c" + std::to_string(column), arrow::int32(), true)); + columns.push_back(std::make_shared<arrow::ChunkedArray>(values)); + } + const auto table = arrow::Table::Make(arrow::schema(std::move(fields)), std::move(columns)); + + const auto output_result = arrow::io::FileOutputStream::Open(temporary_path); + if (!output_result.ok()) { + throw std::runtime_error(output_result.status().ToString()); + } + const auto output = *output_result; + ::parquet::WriterProperties::Builder properties; + properties.version(::parquet::ParquetVersion::PARQUET_2_6); + properties.data_page_version(::parquet::ParquetDataPageVersion::V2); + properties.compression(::parquet::Compression::UNCOMPRESSED); + properties.disable_statistics(); + if (scenario.encoding == Encoding::DICTIONARY) { + properties.enable_dictionary(); + } else { + properties.disable_dictionary(); + properties.encoding(file_encoding(scenario.encoding)); + } + PARQUET_THROW_NOT_OK(::parquet::arrow::WriteTable(*table, arrow::default_memory_pool(), output, + READER_ROW_GROUP_ROWS, properties.build())); + PARQUET_THROW_NOT_OK(output->Close()); + std::filesystem::rename(temporary_path, path); + verify_fixture_encoding(path, scenario); + return path; +} + +class Int32LessThanExpr final : public VExpr { +public: + Int32LessThanExpr(int column_id, int32_t upper_bound) + : VExpr(std::make_shared<DataTypeUInt8>(), false), + _column_id(column_id), + _upper_bound(upper_bound) {} + + Status execute_column_impl(VExprContext*, const Block* block, const Selector* selector, + size_t count, ColumnPtr& result_column) const override { + DORIS_CHECK(block != nullptr); + const auto& nullable = + assert_cast<const ColumnNullable&>(*block->get_by_position(_column_id).column); + const auto& values = assert_cast<const ColumnInt32&>(nullable.get_nested_column()); + const auto& nulls = nullable.get_null_map_data(); + auto result = ColumnUInt8::create(count, 0); + auto& matches = result->get_data(); + for (size_t row = 0; row < count; ++row) { + const size_t input_row = selector == nullptr ? row : (*selector)[row]; + matches[row] = !nulls[input_row] && values.get_element(input_row) < _upper_bound; + } + result_column = std::move(result); + return Status::OK(); + } + + bool can_execute_on_raw_fixed_values(const DataTypePtr& data_type, + int column_id) const override { + return column_id == _column_id && + remove_nullable(data_type)->get_primitive_type() == TYPE_INT; + } + + Status execute_on_raw_fixed_values(const uint8_t* values, size_t num_values, size_t value_width, + const DataTypePtr&, int column_id, + uint8_t* matches) const override { + DORIS_CHECK(column_id == _column_id); + DORIS_CHECK(value_width == sizeof(int32_t)); + const auto* typed_values = reinterpret_cast<const int32_t*>(values); + for (size_t row = 0; row < num_values; ++row) { + matches[row] &= typed_values[row] < _upper_bound; + } + return Status::OK(); + } + + bool can_evaluate_zonemap_filter() const override { return true; } + + ZoneMapFilterResult evaluate_zonemap_filter(const ZoneMapEvalContext& ctx) const override { + const auto zone_map = ctx.zone_map(_column_id); + if (zone_map == nullptr) { + return unsupported_zonemap_filter(ctx); + } + if (!zone_map->has_not_null) { + return ZoneMapFilterResult::kNoMatch; + } + const auto upper_bound = Field::create_field<TYPE_INT>(_upper_bound); + return zone_map->min_value >= upper_bound ? ZoneMapFilterResult::kNoMatch + : ZoneMapFilterResult::kMayMatch; + } + + bool can_evaluate_dictionary_filter() const override { return true; } + + ZoneMapFilterResult evaluate_dictionary_filter( + const DictionaryEvalContext& ctx) const override { + const auto* dictionary = ctx.slot(_column_id); + if (dictionary == nullptr) { + return ZoneMapFilterResult::kUnsupported; + } + const auto upper_bound = Field::create_field<TYPE_INT>(_upper_bound); + return std::ranges::any_of(dictionary->values, + [&](const Field& value) { return value < upper_bound; }) + ? ZoneMapFilterResult::kMayMatch + : ZoneMapFilterResult::kNoMatch; + } + + void collect_slot_column_ids(std::set<int>& column_ids) const override { + column_ids.insert(_column_id); + } + + const std::string& expr_name() const override { return _expr_name; } + +private: + const int _column_id; + const int32_t _upper_bound; + const std::string _expr_name = "ParquetBenchmarkInt32LessThan"; +}; + +inline VExprContextSPtr make_predicate(int column_position, int selectivity_percent) { + auto context = VExprContext::create_shared( + std::make_shared<Int32LessThanExpr>(column_position, selectivity_percent)); + context->_prepared = true; + context->_opened = true; + return context; +} + +inline Block make_block(const std::vector<format::ColumnDefinition>& schema) { + Block block; + for (const auto& column : schema) { + block.insert({column.type->create_column(), column.type, column.name}); + } + return block; +} + +struct ReaderSession { + RuntimeState runtime_state {TQueryOptions(), TQueryGlobals()}; + std::unique_ptr<format::parquet::ParquetReader> reader; + std::vector<format::ColumnDefinition> schema; + std::shared_ptr<format::FileScanRequest> request; +}; + +inline std::unique_ptr<ReaderSession> open_reader(const std::filesystem::path& path, + const ReaderScenario& scenario) { + auto session = std::make_unique<ReaderSession>(); + auto properties = std::make_shared<io::FileSystemProperties>(); + properties->system_type = TFileType::FILE_LOCAL; + auto description = std::make_unique<io::FileDescription>(); + description->path = path.string(); + description->file_size = static_cast<int64_t>(std::filesystem::file_size(path)); + description->range_start_offset = 0; + description->range_size = -1; + session->reader = std::make_unique<format::parquet::ParquetReader>(properties, description, + nullptr, nullptr); + throw_if_error(session->reader->init(&session->runtime_state)); + throw_if_error(session->reader->get_schema(&session->schema)); + + session->request = std::make_shared<format::FileScanRequest>(); + format::FileScanRequestBuilder request_builder(session->request.get()); + if (scenario.operation == ReaderOperation::FULL_SCAN) { + for (int column = 0; column < scenario.schema_width; ++column) { + throw_if_error(request_builder.add_non_predicate_column(format::LocalColumnId(column))); + } + } else if (scenario.operation == ReaderOperation::PREDICATE_SCAN) { + const auto predicate_id = format::LocalColumnId(scenario.predicate_position); + throw_if_error(request_builder.add_predicate_column(predicate_id)); + if (scenario.projection == Projection::PREDICATE_ONLY) { + session->request->predicate_only_columns.push_back(predicate_id); + } else { + const int payload = scenario.predicate_position == 0 ? scenario.schema_width - 1 : 0; + throw_if_error( + request_builder.add_non_predicate_column(format::LocalColumnId(payload))); + } + const auto predicate_position = session->request->local_positions.at(predicate_id).value(); + session->request->conjuncts.push_back( + make_predicate(static_cast<int>(predicate_position), scenario.selectivity_percent)); + } else { + throw_if_error(request_builder.add_non_predicate_column(format::LocalColumnId(0))); + if (scenario.schema_width > 1) { + throw_if_error(request_builder.add_non_predicate_column(format::LocalColumnId(1))); + } + } + + if (scenario.operation == ReaderOperation::LIMIT_1) { + session->reader->set_batch_size(1); + } else if (scenario.operation == ReaderOperation::LIMIT_1000) { + session->reader->set_batch_size(1000); + } + throw_if_error(session->reader->open(session->request)); + return session; +} + +inline size_t scan_reader(ReaderSession* session, const ReaderScenario& scenario) { + size_t output_rows = 0; + bool eof = false; + const size_t limit = scenario.operation == ReaderOperation::LIMIT_1 ? 1 + : scenario.operation == ReaderOperation::LIMIT_1000 ? 1000 + : 0; + while (!eof) { + auto block = make_block(session->schema); Review Comment: [P1] Match the production file-block layout in timed scans Production `TableReader` builds one reusable block template sized and ordered by `request.local_positions`; predicate-only and predicate-plus-payload scans therefore have one and two columns. This creates every file-schema column on every timed `get_block` (512 for the widest cases), and predicate file id 511 is represented by `c0` metadata at output position zero. The width/lazy comparisons therefore include hundreds of unrelated allocations and do not exercise the real layout. Build the template from `local_positions` and clear/reuse it like `TableReader`. ########## be/benchmark/parquet/benchmark_parquet_reader.hpp: ########## @@ -0,0 +1,448 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#pragma once + +#include <arrow/api.h> +#include <arrow/io/api.h> +#include <benchmark/benchmark.h> +#include <parquet/api/reader.h> +#include <parquet/arrow/writer.h> + +#include <algorithm> +#include <cstdint> +#include <filesystem> +#include <memory> +#include <mutex> +#include <set> +#include <stdexcept> +#include <string> +#include <vector> + +#include "core/assert_cast.h" +#include "core/block/block.h" +#include "core/column/column_nullable.h" +#include "core/column/column_vector.h" +#include "core/data_type/data_type_nullable.h" +#include "core/data_type/data_type_number.h" +#include "exprs/vexpr.h" +#include "exprs/vexpr_context.h" +#include "format_v2/file_reader.h" +#include "format_v2/parquet/parquet_reader.h" +#include "gen_cpp/Types_types.h" +#include "io/io_common.h" +#include "parquet_benchmark_scenarios.h" +#include "runtime/runtime_state.h" +#include "storage/index/zone_map/zonemap_eval_context.h" +#include "storage/index/zone_map/zonemap_filter_result.h" + +namespace doris::parquet_benchmark { +namespace reader_detail { + +constexpr size_t READER_ROWS = 1UL << 14; +constexpr size_t READER_ROW_GROUP_ROWS = 1UL << 12; + +inline void throw_if_error(const Status& status) { + if (!status.ok()) { + throw std::runtime_error(status.to_string()); + } +} + +inline bool is_null_row(size_t row, int null_percent, Pattern pattern) { + if (null_percent <= 0) { + return false; + } + if (pattern == Pattern::ALTERNATING) { + constexpr size_t NULL_PERIOD = 101; + const size_t null_slots = (static_cast<size_t>(null_percent) * NULL_PERIOD + 99) / 100; + return (row * 37) % NULL_PERIOD < null_slots; + } + constexpr size_t CLUSTER_ROWS = 1024; + return row % CLUSTER_ROWS < CLUSTER_ROWS * static_cast<size_t>(null_percent) / 100; +} + +inline std::shared_ptr<arrow::Array> build_int32_array(int null_percent, Pattern pattern) { + arrow::Int32Builder builder; + PARQUET_THROW_NOT_OK(builder.Reserve(READER_ROWS)); + for (size_t row = 0; row < READER_ROWS; ++row) { + if (is_null_row(row, null_percent, pattern)) { + PARQUET_THROW_NOT_OK(builder.AppendNull()); + } else { + PARQUET_THROW_NOT_OK(builder.Append(static_cast<int32_t>(row % 100))); + } + } + return builder.Finish().ValueOrDie(); +} + +inline ::parquet::Encoding::type file_encoding(Encoding encoding) { + switch (encoding) { + case Encoding::PLAIN: + return ::parquet::Encoding::PLAIN; + case Encoding::BYTE_STREAM_SPLIT: + return ::parquet::Encoding::BYTE_STREAM_SPLIT; + case Encoding::DELTA_BINARY_PACKED: + return ::parquet::Encoding::DELTA_BINARY_PACKED; + case Encoding::DICTIONARY: + return ::parquet::Encoding::RLE_DICTIONARY; + case Encoding::DELTA_LENGTH_BYTE_ARRAY: + return ::parquet::Encoding::DELTA_LENGTH_BYTE_ARRAY; + case Encoding::DELTA_BYTE_ARRAY: + return ::parquet::Encoding::DELTA_BYTE_ARRAY; + } + throw std::logic_error("unknown Parquet benchmark encoding"); +} + +inline std::string fixture_name(const ReaderScenario& scenario) { + return "v2_" + to_string(scenario.encoding) + "_null" + std::to_string(scenario.null_percent) + + "_" + to_string(scenario.null_pattern) + "_w" + std::to_string(scenario.schema_width) + + "_p" + std::to_string(scenario.predicate_position) + ".parquet"; +} + +inline void verify_fixture_encoding(const std::filesystem::path& path, + const ReaderScenario& scenario) { + auto reader = ::parquet::ParquetFileReader::OpenFile(path.string(), false); + const auto metadata = reader->metadata(); + if (metadata->num_rows() != static_cast<int64_t>(READER_ROWS) || + metadata->num_columns() != scenario.schema_width) { + throw std::runtime_error("Parquet benchmark fixture has unexpected shape: " + + path.string()); + } + const auto expected = file_encoding(scenario.encoding); + for (int row_group = 0; row_group < metadata->num_row_groups(); ++row_group) { + for (int column = 0; column < metadata->num_columns(); ++column) { + const auto encodings = metadata->RowGroup(row_group)->ColumnChunk(column)->encodings(); + if (std::ranges::find(encodings, expected) == encodings.end()) { + throw std::runtime_error("Parquet benchmark fixture did not use " + + to_string(scenario.encoding) + + " encoding: " + path.string()); + } + } + } +} + +inline std::filesystem::path ensure_fixture(const ReaderScenario& scenario) { + static std::mutex fixture_mutex; + const auto directory = + std::filesystem::temp_directory_path() / "doris_parquet_reader_benchmark"; + const auto path = directory / fixture_name(scenario); + std::lock_guard guard(fixture_mutex); + if (std::filesystem::exists(path)) { + verify_fixture_encoding(path, scenario); + return path; + } + + std::filesystem::create_directories(directory); + const auto temporary_path = path.string() + ".tmp"; + std::filesystem::remove(temporary_path); + const auto values = build_int32_array(scenario.null_percent, scenario.null_pattern); + std::vector<std::shared_ptr<arrow::Field>> fields; + std::vector<std::shared_ptr<arrow::ChunkedArray>> columns; + fields.reserve(scenario.schema_width); + columns.reserve(scenario.schema_width); + for (int column = 0; column < scenario.schema_width; ++column) { + fields.push_back(arrow::field("c" + std::to_string(column), arrow::int32(), true)); + columns.push_back(std::make_shared<arrow::ChunkedArray>(values)); + } + const auto table = arrow::Table::Make(arrow::schema(std::move(fields)), std::move(columns)); + + const auto output_result = arrow::io::FileOutputStream::Open(temporary_path); + if (!output_result.ok()) { + throw std::runtime_error(output_result.status().ToString()); + } + const auto output = *output_result; + ::parquet::WriterProperties::Builder properties; + properties.version(::parquet::ParquetVersion::PARQUET_2_6); + properties.data_page_version(::parquet::ParquetDataPageVersion::V2); + properties.compression(::parquet::Compression::UNCOMPRESSED); + properties.disable_statistics(); + if (scenario.encoding == Encoding::DICTIONARY) { + properties.enable_dictionary(); + } else { + properties.disable_dictionary(); + properties.encoding(file_encoding(scenario.encoding)); + } + PARQUET_THROW_NOT_OK(::parquet::arrow::WriteTable(*table, arrow::default_memory_pool(), output, + READER_ROW_GROUP_ROWS, properties.build())); + PARQUET_THROW_NOT_OK(output->Close()); + std::filesystem::rename(temporary_path, path); + verify_fixture_encoding(path, scenario); + return path; +} + +class Int32LessThanExpr final : public VExpr { +public: + Int32LessThanExpr(int column_id, int32_t upper_bound) + : VExpr(std::make_shared<DataTypeUInt8>(), false), + _column_id(column_id), + _upper_bound(upper_bound) {} + + Status execute_column_impl(VExprContext*, const Block* block, const Selector* selector, + size_t count, ColumnPtr& result_column) const override { + DORIS_CHECK(block != nullptr); + const auto& nullable = + assert_cast<const ColumnNullable&>(*block->get_by_position(_column_id).column); + const auto& values = assert_cast<const ColumnInt32&>(nullable.get_nested_column()); + const auto& nulls = nullable.get_null_map_data(); + auto result = ColumnUInt8::create(count, 0); + auto& matches = result->get_data(); + for (size_t row = 0; row < count; ++row) { + const size_t input_row = selector == nullptr ? row : (*selector)[row]; + matches[row] = !nulls[input_row] && values.get_element(input_row) < _upper_bound; + } + result_column = std::move(result); + return Status::OK(); + } + + bool can_execute_on_raw_fixed_values(const DataTypePtr& data_type, + int column_id) const override { + return column_id == _column_id && + remove_nullable(data_type)->get_primitive_type() == TYPE_INT; + } + + Status execute_on_raw_fixed_values(const uint8_t* values, size_t num_values, size_t value_width, + const DataTypePtr&, int column_id, + uint8_t* matches) const override { + DORIS_CHECK(column_id == _column_id); + DORIS_CHECK(value_width == sizeof(int32_t)); + const auto* typed_values = reinterpret_cast<const int32_t*>(values); + for (size_t row = 0; row < num_values; ++row) { + matches[row] &= typed_values[row] < _upper_bound; + } + return Status::OK(); + } + + bool can_evaluate_zonemap_filter() const override { return true; } + + ZoneMapFilterResult evaluate_zonemap_filter(const ZoneMapEvalContext& ctx) const override { + const auto zone_map = ctx.zone_map(_column_id); + if (zone_map == nullptr) { + return unsupported_zonemap_filter(ctx); + } + if (!zone_map->has_not_null) { + return ZoneMapFilterResult::kNoMatch; + } + const auto upper_bound = Field::create_field<TYPE_INT>(_upper_bound); + return zone_map->min_value >= upper_bound ? ZoneMapFilterResult::kNoMatch + : ZoneMapFilterResult::kMayMatch; + } + + bool can_evaluate_dictionary_filter() const override { return true; } + + ZoneMapFilterResult evaluate_dictionary_filter( + const DictionaryEvalContext& ctx) const override { + const auto* dictionary = ctx.slot(_column_id); + if (dictionary == nullptr) { + return ZoneMapFilterResult::kUnsupported; + } + const auto upper_bound = Field::create_field<TYPE_INT>(_upper_bound); + return std::ranges::any_of(dictionary->values, + [&](const Field& value) { return value < upper_bound; }) + ? ZoneMapFilterResult::kMayMatch + : ZoneMapFilterResult::kNoMatch; + } + + void collect_slot_column_ids(std::set<int>& column_ids) const override { + column_ids.insert(_column_id); + } + + const std::string& expr_name() const override { return _expr_name; } + +private: + const int _column_id; + const int32_t _upper_bound; + const std::string _expr_name = "ParquetBenchmarkInt32LessThan"; +}; + +inline VExprContextSPtr make_predicate(int column_position, int selectivity_percent) { + auto context = VExprContext::create_shared( + std::make_shared<Int32LessThanExpr>(column_position, selectivity_percent)); + context->_prepared = true; + context->_opened = true; + return context; +} + +inline Block make_block(const std::vector<format::ColumnDefinition>& schema) { + Block block; + for (const auto& column : schema) { + block.insert({column.type->create_column(), column.type, column.name}); + } + return block; +} + +struct ReaderSession { + RuntimeState runtime_state {TQueryOptions(), TQueryGlobals()}; + std::unique_ptr<format::parquet::ParquetReader> reader; + std::vector<format::ColumnDefinition> schema; + std::shared_ptr<format::FileScanRequest> request; +}; + +inline std::unique_ptr<ReaderSession> open_reader(const std::filesystem::path& path, + const ReaderScenario& scenario) { + auto session = std::make_unique<ReaderSession>(); + auto properties = std::make_shared<io::FileSystemProperties>(); + properties->system_type = TFileType::FILE_LOCAL; + auto description = std::make_unique<io::FileDescription>(); + description->path = path.string(); + description->file_size = static_cast<int64_t>(std::filesystem::file_size(path)); + description->range_start_offset = 0; + description->range_size = -1; + session->reader = std::make_unique<format::parquet::ParquetReader>(properties, description, + nullptr, nullptr); + throw_if_error(session->reader->init(&session->runtime_state)); + throw_if_error(session->reader->get_schema(&session->schema)); + + session->request = std::make_shared<format::FileScanRequest>(); + format::FileScanRequestBuilder request_builder(session->request.get()); + if (scenario.operation == ReaderOperation::FULL_SCAN) { + for (int column = 0; column < scenario.schema_width; ++column) { + throw_if_error(request_builder.add_non_predicate_column(format::LocalColumnId(column))); + } + } else if (scenario.operation == ReaderOperation::PREDICATE_SCAN) { + const auto predicate_id = format::LocalColumnId(scenario.predicate_position); + throw_if_error(request_builder.add_predicate_column(predicate_id)); + if (scenario.projection == Projection::PREDICATE_ONLY) { + session->request->predicate_only_columns.push_back(predicate_id); + } else { + const int payload = scenario.predicate_position == 0 ? scenario.schema_width - 1 : 0; + throw_if_error( + request_builder.add_non_predicate_column(format::LocalColumnId(payload))); + } + const auto predicate_position = session->request->local_positions.at(predicate_id).value(); + session->request->conjuncts.push_back( + make_predicate(static_cast<int>(predicate_position), scenario.selectivity_percent)); + } else { + throw_if_error(request_builder.add_non_predicate_column(format::LocalColumnId(0))); + if (scenario.schema_width > 1) { + throw_if_error(request_builder.add_non_predicate_column(format::LocalColumnId(1))); + } + } + + if (scenario.operation == ReaderOperation::LIMIT_1) { + session->reader->set_batch_size(1); + } else if (scenario.operation == ReaderOperation::LIMIT_1000) { + session->reader->set_batch_size(1000); + } + throw_if_error(session->reader->open(session->request)); + return session; +} + +inline size_t scan_reader(ReaderSession* session, const ReaderScenario& scenario) { + size_t output_rows = 0; + bool eof = false; + const size_t limit = scenario.operation == ReaderOperation::LIMIT_1 ? 1 + : scenario.operation == ReaderOperation::LIMIT_1000 ? 1000 + : 0; + while (!eof) { + auto block = make_block(session->schema); + size_t rows = 0; + throw_if_error(session->reader->get_block(&block, &rows, &eof)); + output_rows += rows; + benchmark::DoNotOptimize(block); + if (scenario.operation == ReaderOperation::OPEN_TO_FIRST_BLOCK) { + break; + } + if (limit != 0 && output_rows >= limit) { + break; + } + } + return output_rows; +} + +inline size_t raw_rows_per_iteration(const ReaderScenario& scenario) { + if (scenario.operation == ReaderOperation::LIMIT_1) { + return 1; + } + if (scenario.operation == ReaderOperation::LIMIT_1000) { + return 1000; + } + if (scenario.operation == ReaderOperation::OPEN_TO_FIRST_BLOCK) { + return format::parquet::ParquetScanScheduler::DEFAULT_READ_BATCH_SIZE; + } + return READER_ROWS; +} + +inline int projected_columns(const ReaderScenario& scenario) { + if (scenario.operation == ReaderOperation::FULL_SCAN) { + return scenario.schema_width; + } + if (scenario.operation == ReaderOperation::PREDICATE_SCAN && + scenario.projection == Projection::PREDICATE_ONLY) { + return 1; + } + return std::min(2, scenario.schema_width); +} + +inline void run_reader(benchmark::State& state, ReaderScenario scenario) { + std::filesystem::path fixture; + try { + fixture = ensure_fixture(scenario); + size_t selected_rows = 0; + for (auto _ : state) { + if (scenario.operation != ReaderOperation::OPEN_TO_FIRST_BLOCK) { + state.PauseTiming(); + } + auto session = open_reader(fixture, scenario); + if (scenario.operation != ReaderOperation::OPEN_TO_FIRST_BLOCK) { + state.ResumeTiming(); + } + selected_rows = scan_reader(session.get(), scenario); Review Comment: [P1] Reject semantically incorrect reader samples This accepts any OK scan as valid, discards all block contents, and extrapolates the last returned row count across the benchmark. A predicate/null/decoder regression can therefore return the wrong rows (or the right count with wrong values) and still be reported as a successful performance sample. Run one untimed fresh-reader oracle per case that checks the exact expected count and deterministic checksums for every semantically returned projected output before entering the timed loop. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
