This is an automated email from the ASF dual-hosted git repository.
lxy-9602 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new c3290904 feat(benchmark): add format-level Parquet read/write
micro-benchmarks (#232)
c3290904 is described below
commit c3290904d46b5bb876fa404f52192aa9de8dae23
Author: Nicholas Jiang <[email protected]>
AuthorDate: Wed Aug 26 17:22:32 2026 +0800
feat(benchmark): add format-level Parquet read/write micro-benchmarks (#232)
---
benchmark/CMakeLists.txt | 31 +
benchmark/parquet_format_benchmark.cpp | 1444 +++++++++++++++++++++++++++
benchmark/parquet_format_benchmark_test.cpp | 533 ++++++++++
docs/source/examples/benchmark.rst | 112 ++-
4 files changed, 2116 insertions(+), 4 deletions(-)
diff --git a/benchmark/CMakeLists.txt b/benchmark/CMakeLists.txt
index 375e9896..0d4dc28c 100644
--- a/benchmark/CMakeLists.txt
+++ b/benchmark/CMakeLists.txt
@@ -63,6 +63,20 @@ if(PAIMON_BUILD_BENCHMARKS)
${PAIMON_BENCHMARK_LINK_TOOLCHAIN}
EXTRA_INCLUDES
${CMAKE_SOURCE_DIR})
+
+ add_paimon_benchmark(parquet_format_benchmark
+ SOURCES
+ parquet_format_benchmark.cpp
+ STATIC_LINK_LIBS
+ arrow
+ parquet
+ ${PAIMON_BENCHMARK_STATIC_LINK_LIBS}
+ test_utils_static
+ Threads::Threads
+ ${PAIMON_BENCHMARK_PLATFORM_LINK_LIBS}
+ ${PAIMON_BENCHMARK_LINK_TOOLCHAIN}
+ EXTRA_INCLUDES
+ ${CMAKE_SOURCE_DIR})
endif()
if(PAIMON_BUILD_TESTS)
@@ -74,4 +88,21 @@ if(PAIMON_BUILD_TESTS)
STATIC_LINK_LIBS
paimon_shared
${GTEST_LINK_TOOLCHAIN})
+
+ # Guards the format-layer assumptions parquet_format_benchmark.cpp is
built on. The benchmark
+ # itself is only compiled under PAIMON_BUILD_BENCHMARKS, which CI does not
set, so this test is
+ # what keeps those assumptions covered.
+ add_paimon_test(parquet_format_benchmark_test
+ SOURCES
+ parquet_format_benchmark_test.cpp
+ EXTRA_INCLUDES
+ ${CMAKE_SOURCE_DIR}
+ STATIC_LINK_LIBS
+ arrow
+ parquet
+ paimon_shared
+ ${PAIMON_LOCAL_FILE_SYSTEM_SHARED_LINK_LIBS}
+ ${PAIMON_PARQUET_FILE_FORMAT_STATIC_LINK_LIBS}
+ test_utils_static
+ ${GTEST_LINK_TOOLCHAIN})
endif()
diff --git a/benchmark/parquet_format_benchmark.cpp
b/benchmark/parquet_format_benchmark.cpp
new file mode 100644
index 00000000..e8d3e002
--- /dev/null
+++ b/benchmark/parquet_format_benchmark.cpp
@@ -0,0 +1,1444 @@
+/*
+ * 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.
+ */
+
+// Format-level micro-benchmarks for the Parquet reader and writer. These drive
+// ParquetWriterBuilder / ParquetFileBatchReader directly, so a format-layer
change can be
+// attributed without the catalog lookup, split planning, merge/sort and
commit that the
+// table-level read_write_benchmark includes. Each case comment says what that
case answers.
+//
+// Every axis - type, cardinality, null density, batch size, encoding,
selectivity - is swept on
+// its own rather than as a combined matrix, because the point is attributing
one change rather
+// than describing a workload.
+//
+// Data is generated outside the timed region and on Arrow's default pool,
because the writer
+// cuts a new row group once its own pool crosses
parquet.writer.max.memory.use.
+//
+// IO goes through the local FileSystem into a temporary directory - the
project has no in-memory
+// FileSystem - so absolute numbers are only meaningful relative to each other.
+
+#include <algorithm>
+#include <atomic>
+#include <chrono>
+#include <cstdint>
+#include <functional>
+#include <map>
+#include <memory>
+#include <mutex>
+#include <optional>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "arrow/api.h"
+#include "arrow/c/abi.h"
+#include "arrow/c/bridge.h"
+#include "arrow/c/helpers.h"
+#include "arrow/util/bit_util.h"
+#include "benchmark/benchmark.h"
+#include "paimon/common/utils/arrow/arrow_input_stream_adapter.h"
+#include "paimon/common/utils/arrow/mem_utils.h"
+#include "paimon/common/utils/arrow/status_utils.h"
+#include "paimon/common/utils/checked_cast.h"
+#include "paimon/common/utils/path_util.h"
+#include "paimon/defs.h"
+#include "paimon/format/format_writer.h"
+#include "paimon/format/parquet/parquet_field_id_converter.h"
+#include "paimon/format/parquet/parquet_file_batch_reader.h"
+#include "paimon/format/parquet/parquet_format_defs.h"
+#include "paimon/format/parquet/parquet_writer_builder.h"
+#include "paimon/fs/file_system.h"
+#include "paimon/memory/memory_pool.h"
+#include "paimon/metrics.h"
+#include "paimon/predicate/literal.h"
+#include "paimon/predicate/predicate.h"
+#include "paimon/predicate/predicate_builder.h"
+#include "paimon/reader/batch_reader.h"
+#include "paimon/result.h"
+#include "paimon/status.h"
+#include "paimon/testing/utils/testharness.h"
+#include "paimon/utils/roaring_bitmap32.h"
+
+namespace {
+
+using ::paimon::ArrowInputStreamAdapter;
+using ::paimon::BatchReader;
+using ::paimon::FieldType;
+using ::paimon::FileStatus;
+using ::paimon::FileSystem;
+using ::paimon::FormatWriter;
+using ::paimon::InputStream;
+using ::paimon::Literal;
+using ::paimon::OutputStream;
+using ::paimon::PathUtil;
+using ::paimon::Predicate;
+using ::paimon::PredicateBuilder;
+using ::paimon::Result;
+using ::paimon::RoaringBitmap32;
+using ::paimon::Status;
+using ::paimon::parquet::ParquetFieldIdConverter;
+using ::paimon::parquet::ParquetFileBatchReader;
+using ::paimon::parquet::ParquetWriterBuilder;
+
+constexpr int64_t kRowsPerFile = 100'000;
+constexpr int64_t kRowsPerBatch = 10'000;
+constexpr int32_t kReadBatchSize = 4096;
+constexpr int32_t kWriteBatchSize = 1024;
+// Small enough that a 100K-row file spans many pages, so page-level pruning
has something to
+// prune. Arrow's page limit is byte-based; a row-count limit is not available
yet.
+constexpr int64_t kPageSizeBytes = 64 * 1024;
+// Four row groups per read fixture, so row-group pruning and page pruning are
both in play.
+constexpr int64_t kRowGroupLength = 25'000;
+constexpr int64_t kStringCardinality = 1'000;
+// Few enough distinct values that arrow keeps the column dictionary-encoded
for the whole file,
+// which is the shape the wide-schema case wants: per-column work small,
per-batch cost visible.
+constexpr int64_t kLowStringCardinality = 10;
+constexpr int32_t kVectorDimension = 16;
+constexpr int32_t kListLength = 4;
+// Same as kListLength, so the MAP and LIST cases differ only in the extra key
leaf.
+constexpr int32_t kMapEntries = kListLength;
+constexpr char kDefaultCompression[] = "zstd";
+// Nulls are placed over a 100-row window, so a requested density is exact,
not statistical.
+constexpr int64_t kNullWindow = 100;
+
+std::shared_ptr<arrow::Field> MakeField(const std::string& name,
+ const
std::shared_ptr<arrow::DataType>& type,
+ int32_t field_id) {
+ return arrow::field(name, type,
+
arrow::KeyValueMetadata::Make({ParquetFieldIdConverter::PARQUET_FIELD_ID},
+
{std::to_string(field_id)}));
+}
+
+std::shared_ptr<arrow::DataType> StructColumnType() {
+ return arrow::struct_({arrow::field("a", arrow::int64()),
arrow::field("b", arrow::utf8())});
+}
+
+std::shared_ptr<arrow::DataType> ListColumnType() {
+ return arrow::list(arrow::int64());
+}
+
+std::shared_ptr<arrow::DataType> VectorColumnType() {
+ return arrow::fixed_size_list(arrow::field("element", arrow::float32(),
/*nullable=*/false),
+ kVectorDimension);
+}
+
+std::shared_ptr<arrow::DataType> MapColumnType() {
+ return arrow::map(arrow::utf8(), arrow::int64());
+}
+
+std::shared_ptr<arrow::DataType> DictionaryStringType() {
+ return arrow::dictionary(arrow::int32(), arrow::utf8());
+}
+
+std::shared_ptr<arrow::DataType> DictionaryInt32Type() {
+ return arrow::dictionary(arrow::int32(), arrow::int32());
+}
+
+// The three-column file the flat read cases scan. `id` is ordered so a range
predicate maps
+// onto a contiguous row range, which is what makes page-index pruning
measurable.
+std::shared_ptr<arrow::Schema> FlatSchema() {
+ return arrow::schema({MakeField("id", arrow::int64(), 0),
MakeField("name", arrow::utf8(), 1),
+ MakeField("amount", arrow::decimal128(18, 4), 2)});
+}
+
+std::shared_ptr<arrow::Schema> DecimalSchema(int32_t precision) {
+ return arrow::schema({MakeField("amount", arrow::decimal128(precision, 4),
0)});
+}
+
+std::shared_ptr<arrow::Schema> DoubleSchema() {
+ return arrow::schema({MakeField("value", arrow::float64(), 0)});
+}
+
+std::shared_ptr<arrow::Schema> NestedSchema() {
+ return arrow::schema(
+ {MakeField("id", arrow::int64(), 0), MakeField("info",
StructColumnType(), 1),
+ MakeField("tags", ListColumnType(), 2), MakeField("embedding",
VectorColumnType(), 3),
+ MakeField("attrs", MapColumnType(), 4)});
+}
+
+// A VECTOR is stored as a Parquet LIST and the reader hands back the file's
own types, so a
+// format-level read asks for the physical type; VectorFileBatchReader
restores the view above.
+std::shared_ptr<arrow::Schema> NestedReadSchema() {
+ return arrow::schema({MakeField("id", arrow::int64(), 0),
+ MakeField("info", StructColumnType(), 1),
+ MakeField("tags", ListColumnType(), 2),
+ MakeField("embedding",
+ arrow::list(arrow::field("element",
arrow::float32(),
+
/*nullable=*/false)),
+ 3),
+ MakeField("attrs", MapColumnType(), 4)});
+}
+
+Result<std::shared_ptr<arrow::Array>> MakeInt64Column(int64_t num_rows,
int64_t offset) {
+ arrow::Int64Builder builder;
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(builder.Reserve(num_rows));
+ for (int64_t i = 0; i < num_rows; ++i) {
+ builder.UnsafeAppend(offset + i);
+ }
+ std::shared_ptr<arrow::Array> array;
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(builder.Finish(&array));
+ return array;
+}
+
+Result<std::shared_ptr<arrow::Array>> MakeDoubleColumn(int64_t num_rows,
int64_t offset) {
+ arrow::DoubleBuilder builder;
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(builder.Reserve(num_rows));
+ for (int64_t i = 0; i < num_rows; ++i) {
+ builder.UnsafeAppend(static_cast<double>(offset + i) * 1.5);
+ }
+ std::shared_ptr<arrow::Array> array;
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(builder.Finish(&array));
+ return array;
+}
+
+Result<std::shared_ptr<arrow::Array>> MakeBooleanColumn(int64_t num_rows,
int64_t offset) {
+ arrow::BooleanBuilder builder;
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(builder.Reserve(num_rows));
+ for (int64_t i = 0; i < num_rows; ++i) {
+ builder.UnsafeAppend(((offset + i) & 1) == 0);
+ }
+ std::shared_ptr<arrow::Array> array;
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(builder.Finish(&array));
+ return array;
+}
+
+// Low cardinality is what arrow dictionary-encodes; high cardinality falls
back to plain.
+Result<std::shared_ptr<arrow::Array>> MakeStringColumn(int64_t num_rows,
int64_t offset,
+ int64_t cardinality) {
+ arrow::StringBuilder builder;
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(builder.Reserve(num_rows));
+ for (int64_t i = 0; i < num_rows; ++i) {
+ const int64_t value = (offset + i) % cardinality;
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(builder.Append("value_" +
std::to_string(value)));
+ }
+ std::shared_ptr<arrow::Array> array;
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(builder.Finish(&array));
+ return array;
+}
+
+// INT32 cycling through `cardinality` distinct values, the flat control for
+// BM_ParquetWrite_DictionaryInt32: same logical values, same cardinality,
same width, so the
+// delta between the two is the dictionary materialization alone.
+Result<std::shared_ptr<arrow::Array>> MakeInt32Column(int64_t num_rows,
int64_t offset,
+ int64_t cardinality) {
+ arrow::Int32Builder builder;
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(builder.Reserve(num_rows));
+ for (int64_t i = 0; i < num_rows; ++i) {
+ builder.UnsafeAppend(static_cast<int32_t>(((offset + i) % cardinality)
* 7));
+ }
+ std::shared_ptr<arrow::Array> array;
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(builder.Finish(&array));
+ return array;
+}
+
+// Dictionary-encoded input: `values->length()` distinct values behind an
int32 index array. Arrow
+// hands the indices straight to Parquet when the value type is binary-like
+// (DictionaryDirectWriteSupported) and densifies them otherwise, so the value
type alone decides
+// whether the writer materializes anything.
+Result<std::shared_ptr<arrow::Array>> MakeDictionaryColumn(
+ const std::shared_ptr<arrow::Array>& values, int64_t num_rows, int64_t
offset) {
+ const int64_t cardinality = values->length();
+ arrow::Int32Builder index_builder;
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(index_builder.Reserve(num_rows));
+ for (int64_t i = 0; i < num_rows; ++i) {
+ index_builder.UnsafeAppend(static_cast<int32_t>((offset + i) %
cardinality));
+ }
+ std::shared_ptr<arrow::Array> indices;
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(index_builder.Finish(&indices));
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Array> array,
+
arrow::DictionaryArray::FromArrays(indices, values));
+ return array;
+}
+
+// STRING values: binary-like, so arrow can write the indices directly.
+Result<std::shared_ptr<arrow::Array>> MakeDictionaryStringColumn(int64_t
num_rows, int64_t offset,
+ int64_t
cardinality) {
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Array> values,
+ MakeStringColumn(cardinality, /*offset=*/0,
cardinality));
+ return MakeDictionaryColumn(values, num_rows, offset);
+}
+
+// INT32 values: is_base_binary_like excludes them, so arrow densifies before
writing. The
+// dictionary holds the same `i * 7` values MakeInt32Column emits inline, so
the two are directly
+// comparable.
+Result<std::shared_ptr<arrow::Array>> MakeDictionaryInt32Column(int64_t
num_rows, int64_t offset,
+ int64_t
cardinality) {
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Array> values,
+ MakeInt32Column(cardinality, /*offset=*/0,
cardinality));
+ return MakeDictionaryColumn(values, num_rows, offset);
+}
+
+// Precision drives the Parquet physical type: <= 9 is INT32, <= 18 is INT64,
larger is
+// FIXED_LEN_BYTE_ARRAY, and the three take different transfer paths on read.
+Result<std::shared_ptr<arrow::Array>> MakeDecimalColumn(int64_t num_rows,
int64_t offset,
+ int32_t precision,
int32_t scale) {
+ arrow::Decimal128Builder builder(arrow::decimal128(precision, scale));
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(builder.Reserve(num_rows));
+ for (int64_t i = 0; i < num_rows; ++i) {
+
PAIMON_RETURN_NOT_OK_FROM_ARROW(builder.Append(arrow::Decimal128(offset + i)));
+ }
+ std::shared_ptr<arrow::Array> array;
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(builder.Finish(&array));
+ return array;
+}
+
+Result<std::shared_ptr<arrow::Array>> MakeStructArray(
+ const arrow::FieldVector& fields, const
std::vector<std::shared_ptr<arrow::Array>>& columns) {
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::StructArray>
array,
+ arrow::StructArray::Make(columns,
fields));
+ return paimon::checked_pointer_cast<arrow::Array>(array);
+}
+
+Result<std::shared_ptr<arrow::Array>> MakeStructColumn(int64_t num_rows,
int64_t offset) {
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Array> a,
MakeInt64Column(num_rows, offset));
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Array> b,
+ MakeStringColumn(num_rows, offset,
kStringCardinality));
+ return MakeStructArray(StructColumnType()->fields(), {a, b});
+}
+
+// Constant element count per row, so the delta against a flat BIGINT column
is the levels.
+Result<std::shared_ptr<arrow::Array>> MakeListColumn(int64_t num_rows, int64_t
offset) {
+ auto value_builder = std::make_shared<arrow::Int64Builder>();
+ arrow::ListBuilder builder(arrow::default_memory_pool(), value_builder,
ListColumnType());
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(builder.Reserve(num_rows));
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(value_builder->Reserve(num_rows *
kListLength));
+ for (int64_t i = 0; i < num_rows; ++i) {
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(builder.Append());
+ for (int32_t j = 0; j < kListLength; ++j) {
+ value_builder->UnsafeAppend(offset + i + j);
+ }
+ }
+ std::shared_ptr<arrow::Array> array;
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(builder.Finish(&array));
+ return array;
+}
+
+// The writer converts VECTOR to a Parquet LIST, so this also covers
ParquetVectorConverter.
+Result<std::shared_ptr<arrow::Array>> MakeVectorColumn(int64_t num_rows,
int64_t offset) {
+ auto value_builder = std::make_shared<arrow::FloatBuilder>();
+ arrow::FixedSizeListBuilder builder(arrow::default_memory_pool(),
value_builder,
+ VectorColumnType());
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(builder.Reserve(num_rows));
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(value_builder->Reserve(num_rows *
kVectorDimension));
+ for (int64_t i = 0; i < num_rows; ++i) {
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(builder.Append());
+ for (int32_t j = 0; j < kVectorDimension; ++j) {
+ value_builder->UnsafeAppend(static_cast<float>((offset + i) %
1024) +
+ static_cast<float>(j));
+ }
+ }
+ std::shared_ptr<arrow::Array> array;
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(builder.Finish(&array));
+ return array;
+}
+
+// A Parquet MAP is a LIST of a two-field STRUCT, so against LIST the
difference is the key leaf.
+Result<std::shared_ptr<arrow::Array>> MakeMapColumn(int64_t num_rows, int64_t
offset,
+ int32_t entries) {
+ auto key_builder = std::make_shared<arrow::StringBuilder>();
+ auto item_builder = std::make_shared<arrow::Int64Builder>();
+ arrow::MapBuilder builder(arrow::default_memory_pool(), key_builder,
item_builder,
+ MapColumnType());
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(builder.Reserve(num_rows));
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(key_builder->Reserve(num_rows * entries));
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(item_builder->Reserve(num_rows * entries));
+ for (int64_t i = 0; i < num_rows; ++i) {
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(builder.Append());
+ for (int32_t j = 0; j < entries; ++j) {
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(key_builder->Append("key_" +
std::to_string(j)));
+ item_builder->UnsafeAppend(offset + i + j);
+ }
+ }
+ std::shared_ptr<arrow::Array> array;
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(builder.Finish(&array));
+ return array;
+}
+
+// Masks `null_pct` percent of the slots null. Doing it after the fact rather
than in every
+// generator keeps the values identical across densities, so the delta between
two settings is the
+// level machinery alone. Only the top level is masked: a STRUCT or LIST row,
not its leaves.
+Result<std::shared_ptr<arrow::Array>> WithNulls(const
std::shared_ptr<arrow::Array>& array,
+ int64_t offset, int64_t
null_pct) {
+ if (null_pct <= 0) {
+ return array;
+ }
+ // Builder output starts at slot 0, so the bitmap below can be indexed by
position.
+ if (array->data()->offset != 0) {
+ return Status::Invalid("WithNulls expects an unsliced array");
+ }
+ const int64_t length = array->length();
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
+ std::shared_ptr<arrow::Buffer> validity,
+ arrow::AllocateEmptyBitmap(length, arrow::default_memory_pool()));
+ int64_t null_count = 0;
+ for (int64_t i = 0; i < length; ++i) {
+ if ((offset + i) % kNullWindow < null_pct) {
+ ++null_count;
+ } else {
+ arrow::bit_util::SetBit(validity->mutable_data(), i);
+ }
+ }
+ std::shared_ptr<arrow::ArrayData> data = array->data()->Copy();
+ data->buffers[0] = std::move(validity);
+ data->SetNullCount(null_count);
+ return arrow::MakeArray(data);
+}
+
+using BatchFactory = std::function<Result<std::shared_ptr<arrow::Array>>(
+ const std::shared_ptr<arrow::Schema>& schema, int64_t offset, int64_t
rows)>;
+
+using ColumnFactory =
+ std::function<Result<std::shared_ptr<arrow::Array>>(int64_t rows, int64_t
offset)>;
+
+ColumnFactory NullableColumnFactory(const ColumnFactory& make_column, int64_t
null_pct) {
+ return [make_column, null_pct](int64_t rows,
+ int64_t offset) ->
Result<std::shared_ptr<arrow::Array>> {
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Array> column,
make_column(rows, offset));
+ return WithNulls(column, offset, null_pct);
+ };
+}
+
+// Lifts a one-column generator into a batch factory for a one-field schema.
+BatchFactory SingleColumnBatch(const ColumnFactory& make_column) {
+ return [make_column](const std::shared_ptr<arrow::Schema>& schema, int64_t
offset,
+ int64_t rows) ->
Result<std::shared_ptr<arrow::Array>> {
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Array> column,
make_column(rows, offset));
+ return MakeStructArray(schema->fields(), {column});
+ };
+}
+
+Result<std::shared_ptr<arrow::Array>> MakeNullableFlatBatch(
+ const std::shared_ptr<arrow::Schema>& schema, int64_t offset, int64_t
rows, int64_t null_pct) {
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Array> ids,
MakeInt64Column(rows, offset));
+ PAIMON_ASSIGN_OR_RAISE(ids, WithNulls(ids, offset, null_pct));
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Array> names,
+ MakeStringColumn(rows, offset, kStringCardinality));
+ PAIMON_ASSIGN_OR_RAISE(names, WithNulls(names, offset, null_pct));
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Array> amounts,
+ MakeDecimalColumn(rows, offset, /*precision=*/18,
/*scale=*/4));
+ PAIMON_ASSIGN_OR_RAISE(amounts, WithNulls(amounts, offset, null_pct));
+ return MakeStructArray(schema->fields(), {ids, names, amounts});
+}
+
+Result<std::shared_ptr<arrow::Array>> MakeFlatBatch(const
std::shared_ptr<arrow::Schema>& schema,
+ int64_t offset, int64_t
rows) {
+ return MakeNullableFlatBatch(schema, offset, rows, /*null_pct=*/0);
+}
+
+// One low-cardinality VARCHAR per field, cheap enough that a wide schema
leaves the per-column
+// setup holding the measurement. The staggered offset keeps the columns from
being identical.
+Result<std::shared_ptr<arrow::Array>> MakeWideBatch(const
std::shared_ptr<arrow::Schema>& schema,
+ int64_t offset, int64_t
rows) {
+ std::vector<std::shared_ptr<arrow::Array>> columns;
+ columns.reserve(schema->num_fields());
+ for (int32_t i = 0; i < schema->num_fields(); ++i) {
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Array> column,
+ MakeStringColumn(rows, offset + i,
kLowStringCardinality));
+ columns.push_back(std::move(column));
+ }
+ return MakeStructArray(schema->fields(), columns);
+}
+
+Result<std::shared_ptr<arrow::Array>> MakeNestedBatch(const
std::shared_ptr<arrow::Schema>& schema,
+ int64_t offset, int64_t
rows) {
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Array> ids,
MakeInt64Column(rows, offset));
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Array> info,
MakeStructColumn(rows, offset));
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Array> tags,
MakeListColumn(rows, offset));
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Array> embedding,
MakeVectorColumn(rows, offset));
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Array> attrs,
+ MakeMapColumn(rows, offset, kMapEntries));
+ return MakeStructArray(schema->fields(), {ids, info, tags, embedding,
attrs});
+}
+
+// Whether every AddBatch call gets its own array or they all share one. It
matters for dictionary
+// input: arrow compares each batch's dictionary against the previous one, so
N distinct-but-equal
+// dictionaries and one dictionary written N times are not the same write. Only
+// BM_ParquetWrite_MemoryThreshold reuses one batch; every other case
generates each batch
+// independently. That is about the arrays, not the values - generators that
cycle with a period
+// dividing the batch size, such as the boolean and the low-cardinality string
ones, produce
+// batches that are equal in value but separate objects.
+enum class BatchReuse { kFresh, kReused };
+
+Result<std::vector<std::shared_ptr<arrow::Array>>> MakeBatches(
+ const std::shared_ptr<arrow::Schema>& schema, const BatchFactory&
make_batch,
+ int64_t rows_per_batch, int64_t total_rows = kRowsPerFile,
+ BatchReuse reuse = BatchReuse::kFresh) {
+ if (reuse == BatchReuse::kReused) {
+ // One array written N times cannot express a short final batch, and
silently writing a
+ // full one instead would put more rows in the file than the reported
metrics divide by.
+ if (total_rows % rows_per_batch != 0) {
+ return Status::Invalid("BatchReuse::kReused needs total_rows
divisible by " +
+ std::to_string(rows_per_batch) + ", got " +
+ std::to_string(total_rows));
+ }
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Array> batch,
+ make_batch(schema, /*offset=*/0,
rows_per_batch));
+ return std::vector<std::shared_ptr<arrow::Array>>(total_rows /
rows_per_batch,
+ std::move(batch));
+ }
+ std::vector<std::shared_ptr<arrow::Array>> batches;
+ for (int64_t offset = 0; offset < total_rows; offset += rows_per_batch) {
+ const int64_t rows = std::min<int64_t>(rows_per_batch, total_rows -
offset);
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Array> batch,
+ make_batch(schema, offset, rows));
+ batches.push_back(std::move(batch));
+ }
+ return batches;
+}
+
+Result<int64_t> WriteParquetFile(const std::shared_ptr<FileSystem>& fs, const
std::string& path,
+ const std::shared_ptr<arrow::Schema>& schema,
+ const
std::vector<std::shared_ptr<arrow::Array>>& batches,
+ const std::map<std::string, std::string>&
options,
+ const std::string& compression) {
+ ParquetWriterBuilder writer_builder(schema, kWriteBatchSize, options);
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<OutputStream> out, fs->Create(path,
/*overwrite=*/true));
+ PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FormatWriter> writer,
+ writer_builder.Build(out, compression));
+ for (const auto& batch : batches) {
+ // AddBatch imports - and so consumes - the C array, so each batch
needs a fresh export.
+ ArrowArray c_array;
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportArray(*batch, &c_array));
+ PAIMON_RETURN_NOT_OK(writer->AddBatch(&c_array));
+ }
+ PAIMON_RETURN_NOT_OK(writer->Finish());
+ PAIMON_RETURN_NOT_OK(out->Close());
+ PAIMON_ASSIGN_OR_RAISE(FileStatus file_status, fs->GetFileStatus(path));
+ return file_status.GetLen();
+}
+
+struct ReadStats {
+ int64_t rows = 0;
+ // Reader-side counterpart of file size: makes pruning and skipping
visible apart from CPU.
+ uint64_t storage_bytes = 0;
+ // Without these a filtered case shows only that it got faster, not that
pruning is why.
+ uint64_t row_groups_total = 0;
+ uint64_t row_groups_after_filter = 0;
+ uint64_t batches = 0;
+};
+
+// Every counter read below is set unconditionally by the reader - the
row-group pair in
+// SetReadSchema, the batch count in NextBatch - so an absent one means the
metric moved or stopped
+// being recorded. Reporting that as a zero would hide the regression behind a
plausible number,
+// so the error propagates and fails the case instead.
+Result<uint64_t> ReadCounter(const std::shared_ptr<paimon::Metrics>& metrics,
+ const std::string& name) {
+ return metrics->GetCounter(name);
+}
+
+Result<ReadStats> ReadParquetFile(const std::shared_ptr<FileSystem>& fs, const
std::string& path,
+ const std::shared_ptr<arrow::Schema>&
read_schema,
+ const std::shared_ptr<Predicate>& predicate,
+ const std::optional<RoaringBitmap32>&
selection_bitmap,
+ const std::map<std::string, std::string>&
options,
+ int32_t batch_size,
+ const std::shared_ptr<arrow::MemoryPool>&
pool) {
+ PAIMON_ASSIGN_OR_RAISE(FileStatus file_status, fs->GetFileStatus(path));
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<InputStream> input, fs->Open(path));
+ auto in_stream = std::make_shared<ArrowInputStreamAdapter>(input,
file_status.GetLen(), pool);
+ // Held separately so the counter outlives the adapter the reader takes
ownership of.
+ std::shared_ptr<std::atomic<uint64_t>> storage_read_bytes =
in_stream->StorageReadBytes();
+ PAIMON_ASSIGN_OR_RAISE(
+ std::unique_ptr<ParquetFileBatchReader> reader,
+ ParquetFileBatchReader::Create(std::move(in_stream), options,
batch_size,
+ /*file_metadata=*/nullptr,
storage_read_bytes, pool,
+ /*hints=*/std::nullopt));
+
+ // SetReadSchema imports the C schema and takes ownership of it.
+ ArrowSchema c_schema;
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportSchema(*read_schema,
&c_schema));
+ PAIMON_RETURN_NOT_OK(reader->SetReadSchema(&c_schema, predicate,
selection_bitmap));
+
+ ReadStats stats;
+ while (true) {
+ PAIMON_ASSIGN_OR_RAISE(BatchReader::ReadBatch batch,
reader->NextBatch());
+ if (BatchReader::IsEofBatch(batch)) {
+ break;
+ }
+ stats.rows += batch.first->length;
+ ArrowArrayRelease(batch.first.get());
+ ArrowSchemaRelease(batch.second.get());
+ }
+
+ std::shared_ptr<paimon::Metrics> metrics = reader->GetReaderMetrics();
+ PAIMON_ASSIGN_OR_RAISE(
+ stats.row_groups_total,
+ ReadCounter(metrics,
paimon::parquet::ParquetMetrics::READ_ROW_GROUPS_TOTAL));
+ PAIMON_ASSIGN_OR_RAISE(
+ stats.row_groups_after_filter,
+ ReadCounter(metrics,
paimon::parquet::ParquetMetrics::READ_ROW_GROUPS_AFTER_FILTER));
+ PAIMON_ASSIGN_OR_RAISE(stats.batches,
+ ReadCounter(metrics,
paimon::parquet::ParquetMetrics::READ_BATCH_COUNT));
+ reader->Close();
+ stats.storage_bytes = storage_read_bytes->load();
+ return stats;
+}
+
+// Row groups the written file actually ended up with. The writer decides that
itself - by row
+// count, or by its pool crossing parquet.writer.max.memory.use - so the
footer is the only
+// reliable source.
+//
+// The write schema is deliberately not used to read back, because for two
cases it does not
+// describe the file. A VECTOR is stored as a LIST, and CollectLeafIndices
branches on the file
+// type, so a FixedSizeList read type against a file LIST is rejected
outright. A dictionary column
+// is stored as its value type; that one survives leaf collection, which
compares nothing for
+// atomic fields, and would fail later against read_data_type_ once a batch is
read. Create()
+// resolves the file's own schema and sets the row-group counters while doing
so, which is all this
+// needs - no read schema, no batch read, and neither trap.
+Result<uint64_t> ReadRowGroupCount(const std::shared_ptr<FileSystem>& fs,
const std::string& path,
+ const std::shared_ptr<arrow::MemoryPool>&
pool) {
+ PAIMON_ASSIGN_OR_RAISE(FileStatus file_status, fs->GetFileStatus(path));
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<InputStream> input, fs->Open(path));
+ auto in_stream = std::make_shared<ArrowInputStreamAdapter>(input,
file_status.GetLen(), pool);
+ PAIMON_ASSIGN_OR_RAISE(
+ std::unique_ptr<ParquetFileBatchReader> reader,
+ ParquetFileBatchReader::Create(std::move(in_stream), /*options=*/{},
kReadBatchSize,
+ /*file_metadata=*/nullptr,
/*storage_read_bytes=*/nullptr,
+ pool, /*hints=*/std::nullopt));
+ PAIMON_ASSIGN_OR_RAISE(uint64_t row_groups,
+ ReadCounter(reader->GetReaderMetrics(),
+
paimon::parquet::ParquetMetrics::READ_ROW_GROUPS_TOTAL));
+ reader->Close();
+ return row_groups;
+}
+
+// Owns a temporary directory plus the shared FileSystem / MemoryPool, removed
on destruction.
+class BenchmarkEnv {
+ public:
+ static Result<std::unique_ptr<BenchmarkEnv>> Create() {
+ std::unique_ptr<paimon::test::UniqueTestDirectory> dir =
+ paimon::test::UniqueTestDirectory::Create();
+ if (!dir) {
+ return Status::IOError("failed to create a temporary benchmark
directory");
+ }
+ return std::unique_ptr<BenchmarkEnv>(new BenchmarkEnv(std::move(dir)));
+ }
+
+ const std::shared_ptr<FileSystem>& fs() const {
+ return fs_;
+ }
+
+ // The reader takes an arrow pool; the writer builder allocates its own
from the paimon pool.
+ const std::shared_ptr<arrow::MemoryPool>& arrow_pool() const {
+ return arrow_pool_;
+ }
+
+ std::string PathOf(const std::string& file_name) const {
+ return PathUtil::JoinPath(dir_->Str(), file_name);
+ }
+
+ private:
+ explicit BenchmarkEnv(std::unique_ptr<paimon::test::UniqueTestDirectory>
dir)
+ : dir_(std::move(dir)),
+ fs_(dir_->GetFileSystem()),
+ arrow_pool_(paimon::GetArrowPool(paimon::GetDefaultPool())) {}
+
+ std::unique_ptr<paimon::test::UniqueTestDirectory> dir_;
+ std::shared_ptr<FileSystem> fs_;
+ std::shared_ptr<arrow::MemoryPool> arrow_pool_;
+};
+
+// A file the read cases scan, built once per configuration; rebuilding it
would dominate.
+class ReadFixture {
+ public:
+ ReadFixture(const std::string& file_name, const
std::shared_ptr<arrow::Schema>& schema,
+ const BatchFactory& make_batch,
+ const std::map<std::string, std::string>& write_options = {}) {
+ status_ = Build(file_name, schema, make_batch, write_options);
+ }
+
+ const Status& status() const {
+ return status_;
+ }
+
+ const std::string& path() const {
+ return path_;
+ }
+
+ const std::shared_ptr<FileSystem>& fs() const {
+ return env_->fs();
+ }
+
+ const std::shared_ptr<arrow::MemoryPool>& arrow_pool() const {
+ return env_->arrow_pool();
+ }
+
+ int64_t file_bytes() const {
+ return file_bytes_;
+ }
+
+ private:
+ Status Build(const std::string& file_name, const
std::shared_ptr<arrow::Schema>& schema,
+ const BatchFactory& make_batch,
+ const std::map<std::string, std::string>& write_options) {
+ PAIMON_ASSIGN_OR_RAISE(env_, BenchmarkEnv::Create());
+ path_ = env_->PathOf(file_name);
+ PAIMON_ASSIGN_OR_RAISE(std::vector<std::shared_ptr<arrow::Array>>
batches,
+ MakeBatches(schema, make_batch, kRowsPerBatch));
+ std::map<std::string, std::string> options = write_options;
+ options[paimon::parquet::PARQUET_PAGE_SIZE] =
std::to_string(kPageSizeBytes);
+ options[paimon::parquet::PARQUET_WRITE_MAX_ROW_GROUP_LENGTH] =
+ std::to_string(kRowGroupLength);
+ PAIMON_ASSIGN_OR_RAISE(file_bytes_, WriteParquetFile(env_->fs(),
path_, schema, batches,
+ options,
kDefaultCompression));
+ return Status::OK();
+ }
+
+ std::unique_ptr<BenchmarkEnv> env_;
+ std::string path_;
+ int64_t file_bytes_ = 0;
+ Status status_;
+};
+
+// Built on first use, so a case excluded by --benchmark_filter never pays to
write its file.
+// google/benchmark may report from a different thread than it registered on,
hence the lock.
+const ReadFixture& GetFixture(const std::string& key,
+ const
std::function<std::unique_ptr<ReadFixture>()>& build) {
+ static std::mutex mutex;
+ static std::map<std::string, std::unique_ptr<ReadFixture>> fixtures;
+ std::lock_guard<std::mutex> guard(mutex);
+ std::unique_ptr<ReadFixture>& fixture = fixtures[key];
+ if (!fixture) {
+ fixture = build();
+ }
+ return *fixture;
+}
+
+const ReadFixture& FlatFixture() {
+ return GetFixture("flat", [] {
+ return std::make_unique<ReadFixture>("flat.parquet", FlatSchema(),
&MakeFlatBatch);
+ });
+}
+
+const ReadFixture& NestedFixture() {
+ return GetFixture("nested", [] {
+ return std::make_unique<ReadFixture>("nested.parquet", NestedSchema(),
&MakeNestedBatch);
+ });
+}
+
+const ReadFixture& NullableFlatFixture(int64_t null_pct) {
+ const std::string key = "flat_nulls_" + std::to_string(null_pct);
+ return GetFixture(key, [key, null_pct] {
+ return std::make_unique<ReadFixture>(
+ key + ".parquet", FlatSchema(),
+ [null_pct](const std::shared_ptr<arrow::Schema>& schema, int64_t
offset, int64_t rows) {
+ return MakeNullableFlatBatch(schema, offset, rows, null_pct);
+ });
+ });
+}
+
+// A one-column file, for read cases that isolate a single decoder.
+const ReadFixture& ColumnFixture(const std::string& key,
+ const std::shared_ptr<arrow::Schema>& schema,
+ const ColumnFactory& make_column) {
+ return GetFixture(key, [key, schema, make_column] {
+ return std::make_unique<ReadFixture>(key + ".parquet", schema,
+ SingleColumnBatch(make_column));
+ });
+}
+
+// The flat fixture only carries precision 18, so without this the
FIXED_LEN_BYTE_ARRAY path that
+// precision 38 takes is written but never read.
+const ReadFixture& DecimalFixture(int32_t precision) {
+ return ColumnFixture("decimal_" + std::to_string(precision),
DecimalSchema(precision),
+ [precision](int64_t rows, int64_t offset) {
+ return MakeDecimalColumn(rows, offset, precision,
/*scale=*/4);
+ });
+}
+
+// The flat schema carries no floating-point column, so a DOUBLE read needs a
file of its own.
+const ReadFixture& DoubleFixture() {
+ return ColumnFixture("double", DoubleSchema(), &MakeDoubleColumn);
+}
+
+// The same data with dictionary encoding off, giving the read side a plain
baseline.
+const ReadFixture& PlainFlatFixture() {
+ return GetFixture("flat_plain", [] {
+ std::map<std::string, std::string> options;
+ options[paimon::parquet::PARQUET_ENABLE_DICTIONARY] = "false";
+ return std::make_unique<ReadFixture>("flat_plain.parquet",
FlatSchema(), &MakeFlatBatch,
+ options);
+ });
+}
+
+// google/benchmark's own main exits 0 whatever happened, so a case that
called SkipWithError
+// prints as skipped while `ctest -L benchmark` still passes. Nothing here
skips on purpose, so
+// the flag every error sets becomes the process exit code. It is recorded
here rather than in a
+// custom reporter because passing one to RunSpecifiedBenchmarks would bypass
+// CreateDefaultDisplayReporter and with it --benchmark_format,
--benchmark_color and
+// --benchmark_counters_tabular.
+std::atomic<bool> g_failed{false};
+
+bool FailBenchmark(::benchmark::State& state, const Status& status) {
+ if (status.ok()) {
+ return false;
+ }
+ g_failed.store(true);
+ state.SkipWithError(status.ToString().c_str());
+ return true;
+}
+
+class Timer {
+ public:
+ double ElapsedNanos() const {
+ return std::chrono::duration<double,
std::nano>(std::chrono::steady_clock::now() - started_)
+ .count();
+ }
+
+ private:
+ std::chrono::steady_clock::time_point started_ =
std::chrono::steady_clock::now();
+};
+
+// google/benchmark reports items/s; ns per row is what the issue asks for, so
the timed region
+// is also measured directly and divided by the row count.
+void ReportRowRate(::benchmark::State& state, int64_t rows, double elapsed_ns)
{
+ state.SetItemsProcessed(static_cast<int64_t>(state.iterations()) * rows);
+ const double total_rows = static_cast<double>(state.iterations()) *
static_cast<double>(rows);
+ if (total_rows > 0) {
+ state.counters["ns_per_row"] = ::benchmark::Counter(elapsed_ns /
total_rows);
+ }
+}
+
+// Bytes per iteration, plus the per-row form that shows a CPU-for-size trade
next to ns_per_row.
+void ReportBytes(::benchmark::State& state, const std::string& name, int64_t
bytes, int64_t rows) {
+ state.counters[name] = ::benchmark::Counter(static_cast<double>(bytes));
+ if (rows > 0) {
+ state.counters["bytes_per_row"] =
+ ::benchmark::Counter(static_cast<double>(bytes) /
static_cast<double>(rows));
+ }
+}
+
+// Everything the writer needs per file - properties, output stream, schema
conversion, footer -
+// is inside the timed region, because that is what a caller pays per data
file.
+void RunWriteBenchmark(::benchmark::State& state, const
std::shared_ptr<arrow::Schema>& schema,
+ const BatchFactory& make_batch, int64_t rows_per_batch,
+ const std::map<std::string, std::string>& options,
+ const std::string& compression, int64_t total_rows =
kRowsPerFile,
+ BatchReuse reuse = BatchReuse::kFresh) {
+ Result<std::unique_ptr<BenchmarkEnv>> env = BenchmarkEnv::Create();
+ if (FailBenchmark(state, env.status())) {
+ return;
+ }
+ Result<std::vector<std::shared_ptr<arrow::Array>>> batches =
+ MakeBatches(schema, make_batch, rows_per_batch, total_rows, reuse);
+ if (FailBenchmark(state, batches.status())) {
+ return;
+ }
+ const std::string path = env.value()->PathOf("write_case.parquet");
+
+ int64_t file_bytes = 0;
+ Timer timer;
+ for (auto _ : state) {
+ Result<int64_t> written = WriteParquetFile(env.value()->fs(), path,
schema, batches.value(),
+ options, compression);
+ if (FailBenchmark(state, written.status())) {
+ return;
+ }
+ file_bytes = written.value();
+ }
+ // Captured before the read-back below, which reopens the file and parses
its footer. Reading
+ // the timer after would put that inside ns_per_row but not inside
google/benchmark's own
+ // real_time, leaving the two disagreeing by a fixed amount that matters
at low iteration
+ // counts.
+ const double elapsed_ns = timer.ElapsedNanos();
+
+ // Read back rather than computed: with a byte-triggered flush the count
is not predictable
+ // from the arguments, and where a row-count limit does make it
predictable, reporting the real
+ // number is what would catch the prediction being wrong.
+ Result<uint64_t> row_groups =
+ ReadRowGroupCount(env.value()->fs(), path, env.value()->arrow_pool());
+ if (FailBenchmark(state, row_groups.status())) {
+ return;
+ }
+
+ ReportRowRate(state, total_rows, elapsed_ns);
+ ReportBytes(state, "file_bytes", file_bytes, total_rows);
+ state.counters["batches"] = ::benchmark::Counter(
+ static_cast<double>((total_rows + rows_per_batch - 1) /
rows_per_batch));
+ state.counters["row_groups"] =
::benchmark::Counter(static_cast<double>(row_groups.value()));
+}
+
+// Single-column variant: one column isolates one encoder.
+void RunColumnWriteBenchmark(::benchmark::State& state, const
std::shared_ptr<arrow::Field>& field,
+ const ColumnFactory& make_column, int64_t
rows_per_batch,
+ const std::map<std::string, std::string>& options,
+ int64_t total_rows = kRowsPerFile,
+ BatchReuse reuse = BatchReuse::kFresh) {
+ RunWriteBenchmark(state, arrow::schema({field}),
SingleColumnBatch(make_column), rows_per_batch,
+ options, kDefaultCompression, total_rows, reuse);
+}
+
+void BM_ParquetWrite_Int64(::benchmark::State& state) {
+ RunColumnWriteBenchmark(state, MakeField("id", arrow::int64(), 0),
&MakeInt64Column,
+ kRowsPerBatch, /*options=*/{});
+}
+
+void BM_ParquetWrite_Double(::benchmark::State& state) {
+ RunColumnWriteBenchmark(state, MakeField("value", arrow::float64(), 0),
&MakeDoubleColumn,
+ kRowsPerBatch, /*options=*/{});
+}
+
+void BM_ParquetWrite_Boolean(::benchmark::State& state) {
+ RunColumnWriteBenchmark(state, MakeField("flag", arrow::boolean(), 0),
&MakeBooleanColumn,
+ kRowsPerBatch, /*options=*/{});
+}
+
+void BM_ParquetWrite_String(::benchmark::State& state) {
+ const int64_t cardinality = state.range(0);
+ RunColumnWriteBenchmark(state, MakeField("name", arrow::utf8(), 0),
+ [cardinality](int64_t rows, int64_t offset) {
+ return MakeStringColumn(rows, offset,
cardinality);
+ },
+ kRowsPerBatch, /*options=*/{});
+}
+
+// arg: number of distinct values. The flat control for
BM_ParquetWrite_DictionaryInt32.
+void BM_ParquetWrite_FlatInt32(::benchmark::State& state) {
+ const int64_t cardinality = state.range(0);
+ RunColumnWriteBenchmark(state, MakeField("value", arrow::int32(), 0),
+ [cardinality](int64_t rows, int64_t offset) {
+ return MakeInt32Column(rows, offset,
cardinality);
+ },
+ kRowsPerBatch, /*options=*/{});
+}
+
+// arg: dictionary cardinality. Same values as BM_ParquetWrite_String at the
same cardinality, but
+// handed to the writer already dictionary-encoded - nothing in
ParquetWriterBuilder rejects a
+// dictionary arrow type, so the writer does reach this path - and the pair
isolates what it saves
+// when it can pass indices through instead of materializing every value.
+void BM_ParquetWrite_DictionaryString(::benchmark::State& state) {
+ const int64_t cardinality = state.range(0);
+ RunColumnWriteBenchmark(state, MakeField("name", DictionaryStringType(),
0),
+ [cardinality](int64_t rows, int64_t offset) {
+ return MakeDictionaryStringColumn(rows,
offset, cardinality);
+ },
+ kRowsPerBatch, /*options=*/{});
+}
+
+// The same axis on an INTEGER dictionary, which arrow cannot direct-write -
is_base_binary_like
+// excludes int32, so it densifies first. Its baseline is
BM_ParquetWrite_FlatInt32 at the same
+// cardinality, not the String case: only the flat INT32 control holds value,
width and encoding
+// fixed, so only that delta is the materialization cost.
+void BM_ParquetWrite_DictionaryInt32(::benchmark::State& state) {
+ const int64_t cardinality = state.range(0);
+ RunColumnWriteBenchmark(state, MakeField("value", DictionaryInt32Type(),
0),
+ [cardinality](int64_t rows, int64_t offset) {
+ return MakeDictionaryInt32Column(rows, offset,
cardinality);
+ },
+ kRowsPerBatch, /*options=*/{});
+}
+
+// Same data with dictionary encoding turned off, as an encoding baseline.
+void BM_ParquetWrite_StringNoDictionary(::benchmark::State& state) {
+ const int64_t cardinality = state.range(0);
+ std::map<std::string, std::string> options;
+ options[paimon::parquet::PARQUET_ENABLE_DICTIONARY] = "false";
+ RunColumnWriteBenchmark(
+ state, MakeField("name", arrow::utf8(), 0),
+ [cardinality](int64_t rows, int64_t offset) {
+ return MakeStringColumn(rows, offset, cardinality);
+ },
+ kRowsPerBatch, options);
+}
+
+void BM_ParquetWrite_Decimal(::benchmark::State& state) {
+ const auto precision = static_cast<int32_t>(state.range(0));
+ RunColumnWriteBenchmark(state, MakeField("amount",
arrow::decimal128(precision, 4), 0),
+ [precision](int64_t rows, int64_t offset) {
+ return MakeDecimalColumn(rows, offset,
precision, /*scale=*/4);
+ },
+ kRowsPerBatch, /*options=*/{});
+}
+
+void BM_ParquetWrite_Struct(::benchmark::State& state) {
+ RunColumnWriteBenchmark(state, MakeField("info", StructColumnType(), 0),
&MakeStructColumn,
+ kRowsPerBatch, /*options=*/{});
+}
+
+void BM_ParquetWrite_List(::benchmark::State& state) {
+ RunColumnWriteBenchmark(state, MakeField("tags", ListColumnType(), 0),
&MakeListColumn,
+ kRowsPerBatch, /*options=*/{});
+}
+
+void BM_ParquetWrite_Vector(::benchmark::State& state) {
+ RunColumnWriteBenchmark(state, MakeField("embedding", VectorColumnType(),
0), &MakeVectorColumn,
+ kRowsPerBatch, /*options=*/{});
+}
+
+// arg: entries per map row.
+void BM_ParquetWrite_Map(::benchmark::State& state) {
+ const auto entries = static_cast<int32_t>(state.range(0));
+ RunColumnWriteBenchmark(
+ state, MakeField("attrs", MapColumnType(), 0),
+ [entries](int64_t rows, int64_t offset) { return MakeMapColumn(rows,
offset, entries); },
+ kRowsPerBatch, /*options=*/{});
+}
+
+// arg: percentage of nulls. Every field is nullable, so the column is
`optional` and definition
+// levels are written at every density including zero; the sweep moves what
arrow's null-free fast
+// path is worth. BIGINT is the narrowest column, so levels are the largest
share of what is left.
+void BM_ParquetWrite_Nulls(::benchmark::State& state) {
+ const int64_t null_pct = state.range(0);
+ RunColumnWriteBenchmark(state, MakeField("id", arrow::int64(), 0),
+ NullableColumnFactory(&MakeInt64Column, null_pct),
kRowsPerBatch,
+ /*options=*/{});
+}
+
+// arg: rows per AddBatch call, at a fixed total row count. Smaller batches
mean the same data
+// carries more per-batch fixed cost, and the file-level setup is amortized
over more calls.
+void BM_ParquetWrite_BatchSize(::benchmark::State& state) {
+ RunColumnWriteBenchmark(state, MakeField("id", arrow::int64(), 0),
&MakeInt64Column,
+ state.range(0), /*options=*/{});
+}
+
+// arg: number of columns at a fixed row count. With BM_ParquetWrite_BatchSize
this separates the
+// two ways per-AddBatch work grows - more calls, or more columns per call -
where that work is
+// AddBatch importing the C array plus arrow walking the schema, neither of
which encodes a value.
+void BM_ParquetWrite_ColumnCount(::benchmark::State& state) {
+ const auto columns = static_cast<int32_t>(state.range(0));
+ arrow::FieldVector fields;
+ fields.reserve(columns);
+ for (int32_t i = 0; i < columns; ++i) {
+ fields.push_back(MakeField("c" + std::to_string(i), arrow::utf8(), i));
+ }
+ RunWriteBenchmark(state, arrow::schema(fields), &MakeWideBatch,
kRowsPerBatch, /*options=*/{},
+ kDefaultCompression);
+ state.counters["columns"] =
::benchmark::Counter(static_cast<double>(columns));
+}
+
+// arg: maximum rows per row group, the write-side half of
BM_ParquetRead_Filtered - a row group
+// is the unit the reader prunes, so more of them buys finer pruning at the
cost of footer
+// metadata, flush work and codec material. The byte-triggered counterpart is
+// BM_ParquetWrite_MemoryThreshold.
+void BM_ParquetWrite_RowGroupSize(::benchmark::State& state) {
+ const int64_t row_group_length = state.range(0);
+ std::map<std::string, std::string> options;
+ options[paimon::parquet::PARQUET_WRITE_MAX_ROW_GROUP_LENGTH] =
std::to_string(row_group_length);
+ RunWriteBenchmark(state, FlatSchema(), &MakeFlatBatch, kRowsPerBatch,
options,
+ kDefaultCompression);
+}
+
+// args: the parquet.writer.max.memory.use threshold in KiB, and the number of
AddBatch calls.
+// Unlike every other case here the file is not a fixed kRowsPerFile: batches
are a fixed 10'000
+// rows and the total grows with the batch count, because the question is
whether retained size
+// accumulates across batches until the writer cuts a row group. Holding the
total fixed and
+// shrinking the batch would measure AddBatch granularity instead, which is
what
+// BM_ParquetWrite_BatchSize already does. The input is low-cardinality
dictionary data, where
+// retained and flat size differ most, written as one batch object repeatedly
rather than a fresh
+// one each time so arrow sees the same dictionary rather than N equal ones.
+//
+// Nothing here sets a row-group row limit, so the byte threshold is the only
thing that can
+// trigger a flush. row_groups reports whether it fired at all; if it reads 1
everywhere the sweep
+// measured nothing, and the unit test only covers the mechanism, not this
particular setting.
+void BM_ParquetWrite_MemoryThreshold(::benchmark::State& state) {
+ constexpr int64_t kFlushRowsPerBatch = 10'000;
+ std::map<std::string, std::string> options;
+ options[paimon::parquet::PARQUET_WRITER_MAX_MEMORY_USE] =
std::to_string(state.range(0) * 1024);
+ RunColumnWriteBenchmark(
+ state, MakeField("name", DictionaryStringType(), 0),
+ [](int64_t rows, int64_t offset) {
+ return MakeDictionaryStringColumn(rows, offset,
kLowStringCardinality);
+ },
+ kFlushRowsPerBatch, options, kFlushRowsPerBatch * state.range(1),
BatchReuse::kReused);
+}
+
+// Codec sweep over the mixed flat schema, so the CPU-versus-size trade shows
on data that is not
+// uniformly one type. Levels stay at the ParquetWriterBuilder defaults. The
names are the ones
+// Parquet accepts, not the ones arrow does: "lz4" resolves to arrow's
LZ4_FRAME, which
+// parquet::IsCodecSupported rejects, so the two framings Parquet defines are
spelled out.
+void BM_ParquetWrite_Compression(::benchmark::State& state, const char*
compression) {
+ RunWriteBenchmark(state, FlatSchema(), &MakeFlatBatch, kRowsPerBatch,
/*options=*/{},
+ compression);
+}
+
+// Rows a read case has to materialize for its number to mean anything.
Pruning is not precise, so
+// a filtered case gets a range rather than an exact count. Without the bound,
a fixture that
+// silently stopped producing rows would just look fast.
+struct RowExpectation {
+ int64_t min = kRowsPerFile;
+ int64_t max = kRowsPerFile;
+};
+
+void RunReadBenchmark(::benchmark::State& state, const ReadFixture& fixture,
+ const std::shared_ptr<arrow::Schema>& read_schema,
+ const std::shared_ptr<Predicate>& predicate,
+ const std::optional<RoaringBitmap32>& selection_bitmap,
+ const std::map<std::string, std::string>& options,
int32_t batch_size,
+ const RowExpectation& expected = {}) {
+ if (FailBenchmark(state, fixture.status())) {
+ return;
+ }
+
+ ReadStats stats;
+ Timer timer;
+ for (auto _ : state) {
+ Result<ReadStats> result =
+ ReadParquetFile(fixture.fs(), fixture.path(), read_schema,
predicate, selection_bitmap,
+ options, batch_size, fixture.arrow_pool());
+ if (FailBenchmark(state, result.status())) {
+ return;
+ }
+ stats = result.value();
+ }
+ if (stats.rows < expected.min || stats.rows > expected.max) {
+ FailBenchmark(state, Status::Invalid("read " +
std::to_string(stats.rows) +
+ " rows, expected " +
std::to_string(expected.min) +
+ " to " +
std::to_string(expected.max)));
+ return;
+ }
+
+ const double elapsed_ns = timer.ElapsedNanos();
+ ReportRowRate(state, stats.rows, elapsed_ns);
+ // Normalized by what the file holds, not what this case materialized:
pruning drops the
+ // per-materialized-row numerator and denominator together, so ns_per_row
and bytes_per_row can
+ // rise while the run gets faster. Only a denominator every setting shares
is comparable across
+ // settings that prune by different amounts.
+ const double total_input_rows =
+ static_cast<double>(state.iterations()) *
static_cast<double>(kRowsPerFile);
+ if (total_input_rows > 0) {
+ state.counters["ns_per_input_row"] = ::benchmark::Counter(elapsed_ns /
total_input_rows);
+ state.counters["bytes_per_input_row"] = ::benchmark::Counter(
+ static_cast<double>(stats.storage_bytes) /
static_cast<double>(kRowsPerFile));
+ }
+ ReportBytes(state, "read_bytes",
static_cast<int64_t>(stats.storage_bytes), stats.rows);
+ state.counters["rows_read"] =
::benchmark::Counter(static_cast<double>(stats.rows));
+ state.counters["file_bytes"] =
::benchmark::Counter(static_cast<double>(fixture.file_bytes()));
+ state.counters["row_groups"] =
+ ::benchmark::Counter(static_cast<double>(stats.row_groups_total));
+ state.counters["row_groups_after_filter"] =
+
::benchmark::Counter(static_cast<double>(stats.row_groups_after_filter));
+ state.counters["batches"] =
::benchmark::Counter(static_cast<double>(stats.batches));
+}
+
+void BM_ParquetRead_FullScan(::benchmark::State& state) {
+ RunReadBenchmark(state, FlatFixture(), FlatSchema(), /*predicate=*/nullptr,
+ /*selection_bitmap=*/std::nullopt, /*options=*/{},
kReadBatchSize);
+}
+
+// One column at a time, separating per-column transfer from materializing the
whole row.
+void BM_ParquetRead_Projection(::benchmark::State& state, const char* column) {
+ std::shared_ptr<arrow::Field> field = FlatSchema()->GetFieldByName(column);
+ if (!field) {
+ FailBenchmark(state, Status::Invalid("unknown projection column"));
+ return;
+ }
+ RunReadBenchmark(state, FlatFixture(), arrow::schema({field}),
/*predicate=*/nullptr,
+ /*selection_bitmap=*/std::nullopt, /*options=*/{},
kReadBatchSize);
+}
+
+// args: percentage of rows the predicate keeps, and whether page-level
pruning is on. The `id`
+// column is ordered, so the surviving rows form a prefix; running both
page-index settings makes
+// the pruning gain attributable instead of merely visible.
+void BM_ParquetRead_Filtered(::benchmark::State& state) {
+ const int64_t threshold = kRowsPerFile * state.range(0) / 100;
+ // `id` is ordered, so row-group statistics alone must already discard
every group past the
+ // threshold - that holds with page-index filtering off too. Bounding at
the row-group grain
+ // rather than at kRowsPerFile is what makes a pruning regression fail the
case instead of
+ // quietly reading the whole file.
+ const int64_t row_group_bound =
+ ((threshold + kRowGroupLength - 1) / kRowGroupLength) *
kRowGroupLength;
+ const bool enable_page_index = state.range(1) != 0;
+ std::shared_ptr<Predicate> predicate = PredicateBuilder::LessThan(
+ /*field_index=*/0, /*field_name=*/"id", FieldType::BIGINT,
Literal(threshold));
+ std::map<std::string, std::string> options;
+ options[paimon::parquet::PARQUET_READ_ENABLE_PAGE_INDEX_FILTER] =
+ enable_page_index ? "true" : "false";
+ RunReadBenchmark(state, FlatFixture(), FlatSchema(), predicate,
+ /*selection_bitmap=*/std::nullopt, options,
kReadBatchSize,
+ RowExpectation{threshold, row_group_bound});
+}
+
+// arg: distance between selected rows. The strides straddle the coalesce hole
limit (32 rows by
+// default), which splits this into two regimes: at or below it neighbouring
single-row ranges
+// merge into long spans, so rows_read runs far ahead of selected_rows and
nothing is skipped;
+// above it every row stays its own range and arrow's Skip decodes and
discards each gap instead.
+// Compare read_bytes and ns_per_input_row across the two.
+void BM_ParquetRead_SkipHeavy(::benchmark::State& state) {
+ const int64_t stride = state.range(0);
+ RoaringBitmap32 bitmap;
+ for (int64_t row = 0; row < kRowsPerFile; row += stride) {
+ bitmap.Add(static_cast<int32_t>(row));
+ }
+ const int64_t selected = bitmap.Cardinality();
+ state.counters["selected_rows"] =
::benchmark::Counter(static_cast<double>(selected));
+ RunReadBenchmark(state, FlatFixture(), FlatSchema(),
/*predicate=*/nullptr, bitmap,
+ /*options=*/{}, kReadBatchSize, RowExpectation{selected,
kRowsPerFile});
+}
+
+// arg: percentage of nulls, the read side of BM_ParquetWrite_Nulls:
definition levels have to be
+// decoded back into a validity bitmap, and past some density arrow may do
less work, not more.
+void BM_ParquetRead_Nulls(::benchmark::State& state) {
+ RunReadBenchmark(state, NullableFlatFixture(state.range(0)), FlatSchema(),
+ /*predicate=*/nullptr, /*selection_bitmap=*/std::nullopt,
/*options=*/{},
+ kReadBatchSize);
+}
+
+// The VARCHAR column from a dictionary-encoded file and from a plain-encoded
one. Both are
+// decoded by arrow, so this is arrow's dictionary path against its plain path
on equal values.
+void BM_ParquetRead_Encoding(::benchmark::State& state, bool
enable_dictionary) {
+ std::shared_ptr<arrow::Field> field = FlatSchema()->GetFieldByName("name");
+ RunReadBenchmark(state, enable_dictionary ? FlatFixture() :
PlainFlatFixture(),
+ arrow::schema({field}), /*predicate=*/nullptr,
+ /*selection_bitmap=*/std::nullopt, /*options=*/{},
kReadBatchSize);
+}
+
+// arg: decimal precision, the read side of BM_ParquetWrite_Decimal. Precision
picks the physical
+// type - INT32, INT64 or FIXED_LEN_BYTE_ARRAY, since ParquetWriterBuilder
enables
+// store_decimal_as_integer - and the three take different paths back to
Decimal128Array.
+void BM_ParquetRead_Decimal(::benchmark::State& state) {
+ const auto precision = static_cast<int32_t>(state.range(0));
+ RunReadBenchmark(state, DecimalFixture(precision),
DecimalSchema(precision),
+ /*predicate=*/nullptr, /*selection_bitmap=*/std::nullopt,
/*options=*/{},
+ kReadBatchSize);
+}
+
+void BM_ParquetRead_Double(::benchmark::State& state) {
+ RunReadBenchmark(state, DoubleFixture(), DoubleSchema(),
/*predicate=*/nullptr,
+ /*selection_bitmap=*/std::nullopt, /*options=*/{},
kReadBatchSize);
+}
+
+// arg: rows per NextBatch call. At a fixed row count this turns the per-batch
fixed cost of
+// ParquetFileBatchReader::NextBatch - Validate, the two metrics counters, the
ArrowSchema
+// export - into a number, separated from the per-row decoding cost.
+void BM_ParquetRead_BatchSize(::benchmark::State& state) {
+ RunReadBenchmark(state, FlatFixture(), FlatSchema(), /*predicate=*/nullptr,
+ /*selection_bitmap=*/std::nullopt, /*options=*/{},
+ static_cast<int32_t>(state.range(0)));
+}
+
+// Each nested column costs definition and repetition levels a flat column
does not pay.
+void BM_ParquetRead_Nested(::benchmark::State& state, const char* column) {
+ std::shared_ptr<arrow::Schema> full_schema = NestedReadSchema();
+ std::shared_ptr<arrow::Schema> read_schema = full_schema;
+ if (column != nullptr) {
+ std::shared_ptr<arrow::Field> field =
full_schema->GetFieldByName(column);
+ if (!field) {
+ FailBenchmark(state, Status::Invalid("unknown nested column"));
+ return;
+ }
+ read_schema = arrow::schema({field});
+ }
+ RunReadBenchmark(state, NestedFixture(), read_schema,
/*predicate=*/nullptr,
+ /*selection_bitmap=*/std::nullopt, /*options=*/{},
kReadBatchSize);
+}
+
+} // namespace
+
+BENCHMARK(BM_ParquetWrite_Int64)->Unit(benchmark::kMillisecond)->UseRealTime();
+BENCHMARK(BM_ParquetWrite_Double)->Unit(benchmark::kMillisecond)->UseRealTime();
+BENCHMARK(BM_ParquetWrite_Boolean)->Unit(benchmark::kMillisecond)->UseRealTime();
+BENCHMARK(BM_ParquetWrite_String)
+ ->ArgName("cardinality")
+ ->Arg(10)
+ ->Arg(1000)
+ ->Arg(kRowsPerFile)
+ ->Unit(benchmark::kMillisecond)
+ ->UseRealTime();
+BENCHMARK(BM_ParquetWrite_StringNoDictionary)
+ ->ArgName("cardinality")
+ ->Arg(10)
+ ->Arg(1000)
+ ->Arg(kRowsPerFile)
+ ->Unit(benchmark::kMillisecond)
+ ->UseRealTime();
+BENCHMARK(BM_ParquetWrite_FlatInt32)
+ ->ArgName("cardinality")
+ ->Arg(10)
+ ->Arg(100)
+ ->Arg(1000)
+ ->Arg(10000)
+ ->Unit(benchmark::kMillisecond)
+ ->UseRealTime();
+BENCHMARK(BM_ParquetWrite_DictionaryString)
+ ->ArgName("cardinality")
+ ->Arg(10)
+ ->Arg(100)
+ ->Arg(1000)
+ ->Arg(10000)
+ ->Unit(benchmark::kMillisecond)
+ ->UseRealTime();
+BENCHMARK(BM_ParquetWrite_DictionaryInt32)
+ ->ArgName("cardinality")
+ ->Arg(10)
+ ->Arg(100)
+ ->Arg(1000)
+ ->Arg(10000)
+ ->Unit(benchmark::kMillisecond)
+ ->UseRealTime();
+BENCHMARK(BM_ParquetWrite_Decimal)
+ ->ArgName("precision")
+ ->Arg(9)
+ ->Arg(18)
+ ->Arg(38)
+ ->Unit(benchmark::kMillisecond)
+ ->UseRealTime();
+BENCHMARK(BM_ParquetWrite_Struct)->Unit(benchmark::kMillisecond)->UseRealTime();
+BENCHMARK(BM_ParquetWrite_List)->Unit(benchmark::kMillisecond)->UseRealTime();
+BENCHMARK(BM_ParquetWrite_Vector)->Unit(benchmark::kMillisecond)->UseRealTime();
+BENCHMARK(BM_ParquetWrite_Map)
+ ->ArgName("entries")
+ ->Arg(3)
+ ->Arg(5)
+ ->Arg(10)
+ ->Unit(benchmark::kMillisecond)
+ ->UseRealTime();
+BENCHMARK(BM_ParquetWrite_Nulls)
+ ->ArgName("null_pct")
+ ->Arg(0)
+ ->Arg(20)
+ ->Arg(50)
+ ->Arg(70)
+ ->Arg(100)
+ ->Unit(benchmark::kMillisecond)
+ ->UseRealTime();
+BENCHMARK(BM_ParquetWrite_BatchSize)
+ ->ArgName("rows_per_batch")
+ ->Arg(100)
+ ->Arg(1000)
+ ->Arg(10000)
+ ->Unit(benchmark::kMillisecond)
+ ->UseRealTime();
+BENCHMARK(BM_ParquetWrite_ColumnCount)
+ ->ArgName("columns")
+ ->Arg(1)
+ ->Arg(5)
+ ->Arg(10)
+ ->Arg(20)
+ ->Unit(benchmark::kMillisecond)
+ ->UseRealTime();
+BENCHMARK(BM_ParquetWrite_RowGroupSize)
+ ->ArgName("row_group_rows")
+ ->Arg(5000)
+ ->Arg(25000)
+ ->Arg(kRowsPerFile)
+ ->Unit(benchmark::kMillisecond)
+ ->UseRealTime();
+BENCHMARK(BM_ParquetWrite_MemoryThreshold)
+ ->ArgNames({"max_memory_kib", "batches"})
+ ->Args({512, 50})
+ ->Args({512, 200})
+ ->Args({4096, 200})
+ ->Args({65536, 200})
+ ->Unit(benchmark::kMillisecond)
+ ->UseRealTime();
+BENCHMARK_CAPTURE(BM_ParquetWrite_Compression, none, "none")
+ ->Unit(benchmark::kMillisecond)
+ ->UseRealTime();
+BENCHMARK_CAPTURE(BM_ParquetWrite_Compression, snappy, "snappy")
+ ->Unit(benchmark::kMillisecond)
+ ->UseRealTime();
+BENCHMARK_CAPTURE(BM_ParquetWrite_Compression, gzip, "gzip")
+ ->Unit(benchmark::kMillisecond)
+ ->UseRealTime();
+BENCHMARK_CAPTURE(BM_ParquetWrite_Compression, lz4_raw, "lz4_raw")
+ ->Unit(benchmark::kMillisecond)
+ ->UseRealTime();
+BENCHMARK_CAPTURE(BM_ParquetWrite_Compression, lz4_hadoop, "lz4_hadoop")
+ ->Unit(benchmark::kMillisecond)
+ ->UseRealTime();
+BENCHMARK_CAPTURE(BM_ParquetWrite_Compression, brotli, "brotli")
+ ->Unit(benchmark::kMillisecond)
+ ->UseRealTime();
+BENCHMARK_CAPTURE(BM_ParquetWrite_Compression, zstd, "zstd")
+ ->Unit(benchmark::kMillisecond)
+ ->UseRealTime();
+
+BENCHMARK(BM_ParquetRead_FullScan)->Unit(benchmark::kMillisecond)->UseRealTime();
+BENCHMARK_CAPTURE(BM_ParquetRead_Projection, id, "id")
+ ->Unit(benchmark::kMillisecond)
+ ->UseRealTime();
+BENCHMARK_CAPTURE(BM_ParquetRead_Projection, name, "name")
+ ->Unit(benchmark::kMillisecond)
+ ->UseRealTime();
+BENCHMARK_CAPTURE(BM_ParquetRead_Projection, amount, "amount")
+ ->Unit(benchmark::kMillisecond)
+ ->UseRealTime();
+BENCHMARK(BM_ParquetRead_Filtered)
+ ->ArgNames({"keep_pct", "page_index"})
+ ->Args({1, 1})
+ ->Args({1, 0})
+ ->Args({10, 1})
+ ->Args({10, 0})
+ ->Args({50, 1})
+ ->Args({50, 0})
+ ->Unit(benchmark::kMillisecond)
+ ->UseRealTime();
+BENCHMARK(BM_ParquetRead_SkipHeavy)
+ ->ArgName("stride")
+ ->Arg(8)
+ ->Arg(32)
+ ->Arg(64)
+ ->Arg(512)
+ ->Arg(4096)
+ ->Unit(benchmark::kMillisecond)
+ ->UseRealTime();
+BENCHMARK(BM_ParquetRead_Nulls)
+ ->ArgName("null_pct")
+ ->Arg(0)
+ ->Arg(20)
+ ->Arg(50)
+ ->Arg(70)
+ ->Arg(100)
+ ->Unit(benchmark::kMillisecond)
+ ->UseRealTime();
+BENCHMARK_CAPTURE(BM_ParquetRead_Encoding, dictionary, true)
+ ->Unit(benchmark::kMillisecond)
+ ->UseRealTime();
+BENCHMARK_CAPTURE(BM_ParquetRead_Encoding, plain, false)
+ ->Unit(benchmark::kMillisecond)
+ ->UseRealTime();
+BENCHMARK(BM_ParquetRead_Decimal)
+ ->ArgName("precision")
+ ->Arg(9)
+ ->Arg(18)
+ ->Arg(38)
+ ->Unit(benchmark::kMillisecond)
+ ->UseRealTime();
+BENCHMARK(BM_ParquetRead_Double)->Unit(benchmark::kMillisecond)->UseRealTime();
+BENCHMARK(BM_ParquetRead_BatchSize)
+ ->ArgName("batch_size")
+ ->Arg(512)
+ ->Arg(4096)
+ ->Arg(16384)
+ ->Unit(benchmark::kMillisecond)
+ ->UseRealTime();
+BENCHMARK_CAPTURE(BM_ParquetRead_Nested, info_struct, "info")
+ ->Unit(benchmark::kMillisecond)
+ ->UseRealTime();
+BENCHMARK_CAPTURE(BM_ParquetRead_Nested, tags_list, "tags")
+ ->Unit(benchmark::kMillisecond)
+ ->UseRealTime();
+BENCHMARK_CAPTURE(BM_ParquetRead_Nested, embedding_vector, "embedding")
+ ->Unit(benchmark::kMillisecond)
+ ->UseRealTime();
+BENCHMARK_CAPTURE(BM_ParquetRead_Nested, attrs_map, "attrs")
+ ->Unit(benchmark::kMillisecond)
+ ->UseRealTime();
+BENCHMARK_CAPTURE(BM_ParquetRead_Nested, all_columns, nullptr)
+ ->Unit(benchmark::kMillisecond)
+ ->UseRealTime();
+
+int main(int argc, char** argv) {
+ ::benchmark::Initialize(&argc, argv);
+ if (::benchmark::ReportUnrecognizedArguments(argc, argv)) {
+ return 1;
+ }
+ ::benchmark::RunSpecifiedBenchmarks();
+ ::benchmark::Shutdown();
+ return g_failed.load() ? 1 : 0;
+}
diff --git a/benchmark/parquet_format_benchmark_test.cpp
b/benchmark/parquet_format_benchmark_test.cpp
new file mode 100644
index 00000000..e930449d
--- /dev/null
+++ b/benchmark/parquet_format_benchmark_test.cpp
@@ -0,0 +1,533 @@
+/*
+ * 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.
+ */
+
+// Smoke test for the assumptions parquet_format_benchmark.cpp is built on.
+//
+// The benchmark is only compiled under PAIMON_BUILD_BENCHMARKS, which CI does
not set, so nothing
+// there runs in CI. Every assumption it makes about the format layer - that a
codec name is
+// accepted, that a dictionary-encoded input array can be written, that a
nested or high-precision
+// column survives a round trip, that a predicate and a selection bitmap
return the rows the
+// benchmark asserts on, that the reader metrics it reports exist - is checked
here instead, at a
+// row count small enough to stay a test. A regression in any of them would
otherwise surface as a
+// benchmark that quietly measures the wrong thing.
+
+#include <cstdint>
+#include <map>
+#include <memory>
+#include <optional>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "arrow/api.h"
+#include "arrow/array/concatenate.h"
+#include "arrow/c/abi.h"
+#include "arrow/c/bridge.h"
+#include "arrow/c/helpers.h"
+#include "gtest/gtest.h"
+#include "paimon/common/utils/arrow/arrow_input_stream_adapter.h"
+#include "paimon/common/utils/arrow/mem_utils.h"
+#include "paimon/common/utils/checked_cast.h"
+#include "paimon/common/utils/path_util.h"
+#include "paimon/defs.h"
+#include "paimon/format/format_writer.h"
+#include "paimon/format/parquet/parquet_field_id_converter.h"
+#include "paimon/format/parquet/parquet_file_batch_reader.h"
+#include "paimon/format/parquet/parquet_format_defs.h"
+#include "paimon/format/parquet/parquet_writer_builder.h"
+#include "paimon/fs/file_system.h"
+#include "paimon/memory/memory_pool.h"
+#include "paimon/metrics.h"
+#include "paimon/predicate/literal.h"
+#include "paimon/predicate/predicate.h"
+#include "paimon/predicate/predicate_builder.h"
+#include "paimon/reader/batch_reader.h"
+#include "paimon/result.h"
+#include "paimon/status.h"
+#include "paimon/testing/utils/testharness.h"
+#include "paimon/utils/roaring_bitmap32.h"
+
+namespace paimon::parquet {
+namespace {
+
+// Small enough to stay a test, large enough to span several batches and row
groups.
+constexpr int64_t kRows = 2'000;
+constexpr int32_t kBatchSize = 256;
+constexpr int64_t kRowGroupLength = 500;
+
+std::shared_ptr<arrow::Field> MakeField(const std::string& name,
+ const
std::shared_ptr<arrow::DataType>& type,
+ int32_t field_id) {
+ return arrow::field(name, type,
+
arrow::KeyValueMetadata::Make({ParquetFieldIdConverter::PARQUET_FIELD_ID},
+
{std::to_string(field_id)}));
+}
+
+class ParquetFormatBenchmarkTest : public ::testing::Test {
+ protected:
+ void SetUp() override {
+ dir_ = test::UniqueTestDirectory::Create();
+ ASSERT_TRUE(dir_);
+ fs_ = dir_->GetFileSystem();
+ pool_ = GetArrowPool(GetDefaultPool());
+ }
+
+ std::string PathOf(const std::string& name) const {
+ return PathUtil::JoinPath(dir_->Str(), name);
+ }
+
+ // Writes the same struct array `batch_count` times through the builder
the benchmark uses.
+ // More than one call matters for dictionary input: the benchmark always
writes several
+ // batches, and every batch after the first hands the writer the same
dictionary again.
+ Status Write(const std::string& path, const
std::shared_ptr<arrow::Schema>& schema,
+ const std::shared_ptr<arrow::Array>& batch, const
std::string& compression,
+ const std::map<std::string, std::string>& extra_options = {},
+ int32_t batch_count = 1) {
+ std::map<std::string, std::string> options = extra_options;
+ // emplace, not assignment: a caller that set its own row-group limit
is testing that.
+ options.emplace(PARQUET_WRITE_MAX_ROW_GROUP_LENGTH,
std::to_string(kRowGroupLength));
+ ParquetWriterBuilder writer_builder(schema, kBatchSize, options);
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<OutputStream> out,
+ fs_->Create(path, /*overwrite=*/true));
+ PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FormatWriter> writer,
+ writer_builder.Build(out, compression));
+ for (int32_t i = 0; i < batch_count; ++i) {
+ ArrowArray c_array;
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportArray(*batch,
&c_array));
+ PAIMON_RETURN_NOT_OK(writer->AddBatch(&c_array));
+ }
+ PAIMON_RETURN_NOT_OK(writer->Finish());
+ return out->Close();
+ }
+
+ // Concatenating `array` with itself `times` times, so a multi-batch write
has an expected
+ // value to be compared against.
+ static Result<std::shared_ptr<arrow::Array>> Repeat(const
std::shared_ptr<arrow::Array>& array,
+ int32_t times) {
+ std::vector<std::shared_ptr<arrow::Array>> chunks(times, array);
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Array>
repeated,
+ arrow::Concatenate(chunks));
+ return repeated;
+ }
+
+ static Result<std::shared_ptr<arrow::Array>> MakeDictionary(
+ const std::shared_ptr<arrow::Array>& indices, const
std::shared_ptr<arrow::Array>& values) {
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Array> array,
+
arrow::DictionaryArray::FromArrays(indices, values));
+ return array;
+ }
+
+ struct ReadResult {
+ int64_t rows = 0;
+ uint64_t row_groups_total = 0;
+ uint64_t row_groups_after_filter = 0;
+ uint64_t batches = 0;
+ // Every batch, imported and concatenated. Row counts alone would let
a decoding bug
+ // through, so the tests compare this against what was written.
+ std::shared_ptr<arrow::Array> data;
+ };
+
+ // Reads the file back the way the benchmark does, reporting what the
benchmark reports on.
+ Result<ReadResult> Read(const std::string& path,
+ const std::shared_ptr<arrow::Schema>& read_schema,
+ const std::shared_ptr<Predicate>& predicate =
nullptr,
+ const std::optional<RoaringBitmap32>& selection =
std::nullopt) {
+ PAIMON_ASSIGN_OR_RAISE(FileStatus file_status,
fs_->GetFileStatus(path));
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<InputStream> input,
fs_->Open(path));
+ auto in_stream =
+ std::make_shared<ArrowInputStreamAdapter>(input,
file_status.GetLen(), pool_);
+ PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<ParquetFileBatchReader> reader,
+ ParquetFileBatchReader::Create(
+ std::move(in_stream), /*options=*/{},
kBatchSize,
+ /*file_metadata=*/nullptr,
/*storage_read_bytes=*/nullptr, pool_,
+ /*hints=*/std::nullopt));
+ ArrowSchema c_schema;
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportSchema(*read_schema,
&c_schema));
+ PAIMON_RETURN_NOT_OK(reader->SetReadSchema(&c_schema, predicate,
selection));
+
+ ReadResult result;
+ std::vector<std::shared_ptr<arrow::Array>> chunks;
+ while (true) {
+ PAIMON_ASSIGN_OR_RAISE(BatchReader::ReadBatch batch,
reader->NextBatch());
+ if (BatchReader::IsEofBatch(batch)) {
+ break;
+ }
+ // ImportArray takes ownership of both C structs, so nothing is
released by hand here.
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
+ std::shared_ptr<arrow::Array> chunk,
+ arrow::ImportArray(batch.first.get(), batch.second.get()));
+ result.rows += chunk->length();
+ chunks.push_back(std::move(chunk));
+ }
+ if (!chunks.empty()) {
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(result.data,
arrow::Concatenate(chunks));
+ }
+
+ std::shared_ptr<Metrics> metrics = reader->GetReaderMetrics();
+ PAIMON_ASSIGN_OR_RAISE(result.row_groups_total,
+
metrics->GetCounter(ParquetMetrics::READ_ROW_GROUPS_TOTAL));
+ PAIMON_ASSIGN_OR_RAISE(result.row_groups_after_filter,
+
metrics->GetCounter(ParquetMetrics::READ_ROW_GROUPS_AFTER_FILTER));
+ PAIMON_ASSIGN_OR_RAISE(result.batches,
+
metrics->GetCounter(ParquetMetrics::READ_BATCH_COUNT));
+ reader->Close();
+ return result;
+ }
+
+ static Result<std::shared_ptr<arrow::Array>> Wrap(
+ const std::shared_ptr<arrow::Schema>& schema,
+ const std::vector<std::shared_ptr<arrow::Array>>& columns) {
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::StructArray>
array,
+ arrow::StructArray::Make(columns,
schema->fields()));
+ return checked_pointer_cast<arrow::Array>(array);
+ }
+
+ std::unique_ptr<test::UniqueTestDirectory> dir_;
+ std::shared_ptr<FileSystem> fs_;
+ std::shared_ptr<arrow::MemoryPool> pool_;
+};
+
+// Every codec name the benchmark registers has to be one Parquet accepts.
"lz4" is the trap this
+// guards: it resolves to arrow's LZ4_FRAME, which parquet::IsCodecSupported
rejects, and the
+// failure only shows up once a column chunk is actually written.
+TEST_F(ParquetFormatBenchmarkTest, RegisteredCodecsWrite) {
+ std::shared_ptr<arrow::Schema> schema = arrow::schema({MakeField("id",
arrow::int64(), 0)});
+ arrow::Int64Builder builder;
+ ASSERT_TRUE(builder.Reserve(kRows).ok());
+ for (int64_t i = 0; i < kRows; ++i) {
+ builder.UnsafeAppend(i);
+ }
+ std::shared_ptr<arrow::Array> ids;
+ ASSERT_TRUE(builder.Finish(&ids).ok());
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::Array> batch, Wrap(schema,
{ids}));
+
+ for (const std::string& codec :
+ {"none", "snappy", "gzip", "brotli", "zstd", "lz4_raw",
"lz4_hadoop"}) {
+ const std::string path = PathOf("codec_" + codec + ".parquet");
+ ASSERT_OK(Write(path, schema, batch, codec)) << "codec " << codec;
+ ASSERT_OK_AND_ASSIGN(ReadResult result, Read(path, schema));
+ EXPECT_EQ(kRows, result.rows) << "codec " << codec;
+ ASSERT_TRUE(result.data);
+ EXPECT_TRUE(result.data->Equals(*batch)) << "codec " << codec;
+ }
+
+ // The name the benchmark deliberately does not register still has to be
rejected; if arrow
+ // ever starts accepting it, the comment explaining its absence is stale.
+ EXPECT_FALSE(Write(PathOf("codec_lz4.parquet"), schema, batch,
"lz4").ok());
+}
+
+// A dictionary-encoded input array must reach the writer intact. VARCHAR
takes arrow's direct
+// write path and INT32 gets densified first; both have to produce a readable
file whose logical
+// values match the flat equivalent.
+TEST_F(ParquetFormatBenchmarkTest, DictionaryInputRoundTrip) {
+ constexpr int64_t kCardinality = 8;
+ arrow::Int32Builder index_builder;
+ ASSERT_TRUE(index_builder.Reserve(kRows).ok());
+ for (int64_t i = 0; i < kRows; ++i) {
+ index_builder.UnsafeAppend(static_cast<int32_t>(i % kCardinality));
+ }
+ std::shared_ptr<arrow::Array> indices;
+ ASSERT_TRUE(index_builder.Finish(&indices).ok());
+
+ arrow::StringBuilder string_values;
+ arrow::Int32Builder int_values;
+ for (int64_t i = 0; i < kCardinality; ++i) {
+ ASSERT_TRUE(string_values.Append("value_" + std::to_string(i)).ok());
+ ASSERT_TRUE(int_values.Append(static_cast<int32_t>(i * 7)).ok());
+ }
+ std::shared_ptr<arrow::Array> string_dict;
+ std::shared_ptr<arrow::Array> int_dict;
+ ASSERT_TRUE(string_values.Finish(&string_dict).ok());
+ ASSERT_TRUE(int_values.Finish(&int_dict).ok());
+
+ // The flat arrays the dictionary-encoded input has to decode back to.
+ arrow::StringBuilder flat_strings;
+ arrow::Int32Builder flat_ints;
+ for (int64_t i = 0; i < kRows; ++i) {
+ ASSERT_TRUE(flat_strings.Append("value_" + std::to_string(i %
kCardinality)).ok());
+ ASSERT_TRUE(flat_ints.Append(static_cast<int32_t>((i % kCardinality) *
7)).ok());
+ }
+ std::shared_ptr<arrow::Array> flat_string_column;
+ std::shared_ptr<arrow::Array> flat_int_column;
+ ASSERT_TRUE(flat_strings.Finish(&flat_string_column).ok());
+ ASSERT_TRUE(flat_ints.Finish(&flat_int_column).ok());
+
+ struct Case {
+ const char* name;
+ std::shared_ptr<arrow::Array> dictionary;
+ std::shared_ptr<arrow::DataType> read_type;
+ std::shared_ptr<arrow::Array> flat;
+ };
+ // The benchmark always writes several batches, so the writer sees the
same dictionary more
+ // than once.
+ constexpr int32_t kBatches = 3;
+ for (const Case& c : {Case{"string", string_dict, arrow::utf8(),
flat_string_column},
+ Case{"int32", int_dict, arrow::int32(),
flat_int_column}}) {
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::Array> dictionary_array,
+ MakeDictionary(indices, c.dictionary));
+ std::shared_ptr<arrow::Schema> write_schema =
+ arrow::schema({MakeField("v", dictionary_array->type(), 0)});
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::Array> batch,
+ Wrap(write_schema, {dictionary_array}));
+ const std::string path = PathOf(std::string("dict_") + c.name +
".parquet");
+ ASSERT_OK(Write(path, write_schema, batch, "zstd",
/*extra_options=*/{}, kBatches))
+ << c.name;
+
+ // Parquet has no dictionary type: the column comes back as its value
type either way, and
+ // has to carry the values the flat equivalent would have.
+ std::shared_ptr<arrow::Schema> read_schema =
+ arrow::schema({MakeField("v", c.read_type, 0)});
+ ASSERT_OK_AND_ASSIGN(ReadResult result, Read(path, read_schema));
+ EXPECT_EQ(kRows * kBatches, result.rows) << c.name;
+ ASSERT_TRUE(result.data);
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::Array> expected,
Repeat(c.flat, kBatches));
+ std::shared_ptr<arrow::Array> actual =
+ checked_pointer_cast<arrow::StructArray>(result.data)->field(0);
+ EXPECT_TRUE(actual->Equals(*expected)) << c.name;
+ }
+}
+
+// DECIMAL precision selects the Parquet physical type, and precision 38 is
the only one that
+// reaches FIXED_LEN_BYTE_ARRAY. The benchmark sweeps all three on both sides,
so all three have
+// to survive a round trip with their type intact.
+TEST_F(ParquetFormatBenchmarkTest, DecimalPrecisionRoundTrip) {
+ for (int32_t precision : {9, 18, 38}) {
+ std::shared_ptr<arrow::DataType> type = arrow::decimal128(precision,
4);
+ std::shared_ptr<arrow::Schema> schema =
arrow::schema({MakeField("amount", type, 0)});
+ arrow::Decimal128Builder builder(type);
+ ASSERT_TRUE(builder.Reserve(kRows).ok());
+ for (int64_t i = 0; i < kRows; ++i) {
+ ASSERT_TRUE(builder.Append(arrow::Decimal128(i)).ok());
+ }
+ std::shared_ptr<arrow::Array> amounts;
+ ASSERT_TRUE(builder.Finish(&amounts).ok());
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::Array> batch, Wrap(schema,
{amounts}));
+
+ const std::string path = PathOf("decimal_" + std::to_string(precision)
+ ".parquet");
+ ASSERT_OK(Write(path, schema, batch, "zstd")) << "precision " <<
precision;
+ ASSERT_OK_AND_ASSIGN(ReadResult result, Read(path, schema));
+ EXPECT_EQ(kRows, result.rows) << "precision " << precision;
+ ASSERT_TRUE(result.data);
+ EXPECT_TRUE(result.data->type()->field(0)->type()->Equals(*type))
+ << "precision " << precision;
+ EXPECT_TRUE(result.data->Equals(*batch)) << "precision " << precision;
+ }
+}
+
+// The nested fixture the benchmark reads is only meaningful if LIST and MAP
survive the round
+// trip with the shape the read schema asks for.
+TEST_F(ParquetFormatBenchmarkTest, NestedRoundTrip) {
+ constexpr int32_t kEntries = 4;
+ auto list_values = std::make_shared<arrow::Int64Builder>();
+ arrow::ListBuilder list_builder(arrow::default_memory_pool(), list_values,
+ arrow::list(arrow::int64()));
+ auto key_builder = std::make_shared<arrow::StringBuilder>();
+ auto item_builder = std::make_shared<arrow::Int64Builder>();
+ arrow::MapBuilder map_builder(arrow::default_memory_pool(), key_builder,
item_builder,
+ arrow::map(arrow::utf8(), arrow::int64()));
+ for (int64_t i = 0; i < kRows; ++i) {
+ ASSERT_TRUE(list_builder.Append().ok());
+ ASSERT_TRUE(map_builder.Append().ok());
+ for (int32_t j = 0; j < kEntries; ++j) {
+ ASSERT_TRUE(list_values->Append(i + j).ok());
+ ASSERT_TRUE(key_builder->Append("key_" + std::to_string(j)).ok());
+ ASSERT_TRUE(item_builder->Append(i + j).ok());
+ }
+ }
+ std::shared_ptr<arrow::Array> tags;
+ std::shared_ptr<arrow::Array> attrs;
+ ASSERT_TRUE(list_builder.Finish(&tags).ok());
+ ASSERT_TRUE(map_builder.Finish(&attrs).ok());
+
+ std::shared_ptr<arrow::Schema> schema =
+ arrow::schema({MakeField("tags", arrow::list(arrow::int64()), 0),
+ MakeField("attrs", arrow::map(arrow::utf8(),
arrow::int64()), 1)});
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::Array> batch, Wrap(schema,
{tags, attrs}));
+ const std::string path = PathOf("nested.parquet");
+ ASSERT_OK(Write(path, schema, batch, "zstd"));
+
+ ASSERT_OK_AND_ASSIGN(ReadResult result, Read(path, schema));
+ EXPECT_EQ(kRows, result.rows);
+ ASSERT_TRUE(result.data);
+ EXPECT_TRUE(result.data->Equals(*batch));
+
+ // Each nested column also has to be readable on its own, which is what
the projected nested
+ // cases do.
+ for (const std::string& column : {"tags", "attrs"}) {
+ std::shared_ptr<arrow::Schema> projected =
arrow::schema({schema->GetFieldByName(column)});
+ ASSERT_OK_AND_ASSIGN(ReadResult projected_result, Read(path,
projected));
+ EXPECT_EQ(kRows, projected_result.rows) << "column " << column;
+ }
+}
+
+// The filtered and skip-heavy cases assert on row counts, so those counts
have to mean what the
+// benchmark assumes: a predicate keeps at least the matching rows and no more
than the row groups
+// that could hold them, and a selection bitmap keeps at least the rows it
selected.
+TEST_F(ParquetFormatBenchmarkTest, FilteredAndBitmapRowCounts) {
+ std::shared_ptr<arrow::Schema> schema = arrow::schema({MakeField("id",
arrow::int64(), 0)});
+ arrow::Int64Builder builder;
+ ASSERT_TRUE(builder.Reserve(kRows).ok());
+ for (int64_t i = 0; i < kRows; ++i) {
+ builder.UnsafeAppend(i);
+ }
+ std::shared_ptr<arrow::Array> ids;
+ ASSERT_TRUE(builder.Finish(&ids).ok());
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::Array> batch, Wrap(schema,
{ids}));
+ const std::string path = PathOf("filtered.parquet");
+ ASSERT_OK(Write(path, schema, batch, "zstd"));
+
+ ASSERT_OK_AND_ASSIGN(ReadResult full, Read(path, schema));
+ EXPECT_EQ(kRows, full.rows);
+ EXPECT_EQ(static_cast<uint64_t>(kRows / kRowGroupLength),
full.row_groups_total);
+ EXPECT_EQ(full.row_groups_total, full.row_groups_after_filter);
+ EXPECT_GT(full.batches, 0u);
+
+ // `id` is ordered, so row-group statistics alone must discard everything
past the threshold.
+ // This is the bound BM_ParquetRead_Filtered asserts on.
+ constexpr int64_t kThreshold = 300;
+ std::shared_ptr<Predicate> predicate = PredicateBuilder::LessThan(
+ /*field_index=*/0, /*field_name=*/"id", FieldType::BIGINT,
Literal(kThreshold));
+ ASSERT_OK_AND_ASSIGN(ReadResult filtered, Read(path, schema, predicate));
+ EXPECT_GE(filtered.rows, kThreshold);
+ EXPECT_LE(filtered.rows, kRowGroupLength);
+ EXPECT_LT(filtered.row_groups_after_filter, filtered.row_groups_total);
+
+ // Selection is not precise, so the bitmap is a lower bound on what comes
back.
+ RoaringBitmap32 bitmap;
+ int64_t selected = 0;
+ for (int64_t row = 0; row < kRows; row += 64) {
+ bitmap.Add(static_cast<int32_t>(row));
+ ++selected;
+ }
+ ASSERT_OK_AND_ASSIGN(ReadResult skipped, Read(path, schema,
/*predicate=*/nullptr, bitmap));
+ EXPECT_GE(skipped.rows, selected);
+ EXPECT_LE(skipped.rows, kRows);
+}
+
+// The encoding case compares a dictionary-encoded file against a plain one,
which is only a
+// comparison if both files read back identically.
+TEST_F(ParquetFormatBenchmarkTest, PlainAndDictionaryFilesAgree) {
+ std::shared_ptr<arrow::Schema> schema = arrow::schema({MakeField("name",
arrow::utf8(), 0)});
+ arrow::StringBuilder builder;
+ ASSERT_TRUE(builder.Reserve(kRows).ok());
+ for (int64_t i = 0; i < kRows; ++i) {
+ ASSERT_TRUE(builder.Append("value_" + std::to_string(i % 16)).ok());
+ }
+ std::shared_ptr<arrow::Array> names;
+ ASSERT_TRUE(builder.Finish(&names).ok());
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::Array> batch, Wrap(schema,
{names}));
+
+ const std::string dict_path = PathOf("encoding_dict.parquet");
+ const std::string plain_path = PathOf("encoding_plain.parquet");
+ ASSERT_OK(Write(dict_path, schema, batch, "zstd"));
+ std::map<std::string, std::string> plain_options;
+ plain_options[PARQUET_ENABLE_DICTIONARY] = "false";
+ ASSERT_OK(Write(plain_path, schema, batch, "zstd", plain_options));
+
+ ASSERT_OK_AND_ASSIGN(ReadResult dict_result, Read(dict_path, schema));
+ ASSERT_OK_AND_ASSIGN(ReadResult plain_result, Read(plain_path, schema));
+ EXPECT_EQ(kRows, dict_result.rows);
+ EXPECT_EQ(dict_result.rows, plain_result.rows);
+ EXPECT_EQ(dict_result.row_groups_total, plain_result.row_groups_total);
+ ASSERT_TRUE(dict_result.data);
+ ASSERT_TRUE(plain_result.data);
+ EXPECT_TRUE(dict_result.data->Equals(*batch));
+ EXPECT_TRUE(plain_result.data->Equals(*batch));
+}
+
+// BM_ParquetWrite_MemoryThreshold rests on one assumption: that a small
+// parquet.writer.max.memory.use actually makes ParquetFormatWriter cut extra
row groups. This
+// covers the mechanism on both input shapes that case can present it with,
plain and
+// dictionary-encoded, with the row-count limit raised out of the way so the
byte threshold is the
+// only thing that can flush.
+//
+// It does not cover the benchmark's own 512 KiB setting: reaching that with
cardinality-10
+// dictionary data takes the 500K to 2M rows the benchmark writes, which is
not a unit test. What
+// confirms that setting is the benchmark's own row_groups counter reading
more than 1.
+TEST_F(ParquetFormatBenchmarkTest, MemoryThresholdFlushesRowGroups) {
+ constexpr int32_t kBatches = 8;
+ constexpr int64_t kDictCardinality = 10;
+
+ arrow::StringBuilder plain_builder;
+ ASSERT_TRUE(plain_builder.Reserve(kRows).ok());
+ for (int64_t i = 0; i < kRows; ++i) {
+ ASSERT_TRUE(plain_builder.Append("value_" + std::to_string(i)).ok());
+ }
+ std::shared_ptr<arrow::Array> plain_column;
+ ASSERT_TRUE(plain_builder.Finish(&plain_column).ok());
+
+ // The shape BM_ParquetWrite_MemoryThreshold writes: low-cardinality
dictionary input.
+ arrow::StringBuilder dict_values;
+ for (int64_t i = 0; i < kDictCardinality; ++i) {
+ ASSERT_TRUE(dict_values.Append("value_" + std::to_string(i)).ok());
+ }
+ std::shared_ptr<arrow::Array> dict_value_column;
+ ASSERT_TRUE(dict_values.Finish(&dict_value_column).ok());
+ arrow::Int32Builder dict_indices;
+ ASSERT_TRUE(dict_indices.Reserve(kRows).ok());
+ for (int64_t i = 0; i < kRows; ++i) {
+ dict_indices.UnsafeAppend(static_cast<int32_t>(i % kDictCardinality));
+ }
+ std::shared_ptr<arrow::Array> index_column;
+ ASSERT_TRUE(dict_indices.Finish(&index_column).ok());
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::Array> dict_column,
+ MakeDictionary(index_column, dict_value_column));
+
+ struct Case {
+ const char* name;
+ std::shared_ptr<arrow::Array> column;
+ };
+ for (const Case& c : {Case{"plain", plain_column}, Case{"dictionary",
dict_column}}) {
+ std::shared_ptr<arrow::Schema> write_schema =
+ arrow::schema({MakeField("name", c.column->type(), 0)});
+ // Parquet stores a dictionary column as its value type, so both read
back as UTF8.
+ std::shared_ptr<arrow::Schema> read_schema =
+ arrow::schema({MakeField("name", arrow::utf8(), 0)});
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::Array> batch,
Wrap(write_schema, {c.column}));
+
+ std::map<std::string, std::string> options;
+ options[PARQUET_WRITE_MAX_ROW_GROUP_LENGTH] = std::to_string(kRows *
kBatches * 10);
+
+ options[PARQUET_WRITER_MAX_MEMORY_USE] = std::to_string(uint64_t{8} *
1024);
+ const std::string small_path =
+ PathOf(std::string("threshold_small_") + c.name + ".parquet");
+ ASSERT_OK(Write(small_path, write_schema, batch, "zstd", options,
kBatches)) << c.name;
+ ASSERT_OK_AND_ASSIGN(ReadResult small, Read(small_path, read_schema));
+ EXPECT_EQ(kRows * kBatches, small.rows) << c.name;
+ EXPECT_GT(small.row_groups_total, 1u)
+ << c.name << ": byte threshold never triggered a flush";
+
+ options[PARQUET_WRITER_MAX_MEMORY_USE] =
+ std::to_string(DEFAULT_PARQUET_WRITER_MAX_MEMORY_USE);
+ const std::string large_path =
+ PathOf(std::string("threshold_large_") + c.name + ".parquet");
+ ASSERT_OK(Write(large_path, write_schema, batch, "zstd", options,
kBatches)) << c.name;
+ ASSERT_OK_AND_ASSIGN(ReadResult large, Read(large_path, read_schema));
+ EXPECT_EQ(kRows * kBatches, large.rows) << c.name;
+ EXPECT_EQ(1u, large.row_groups_total) << c.name;
+
+ // Row-group boundaries must not change the data.
+ ASSERT_TRUE(small.data) << c.name;
+ ASSERT_TRUE(large.data) << c.name;
+ EXPECT_TRUE(small.data->Equals(*large.data)) << c.name;
+ }
+}
+
+} // namespace
+} // namespace paimon::parquet
diff --git a/docs/source/examples/benchmark.rst
b/docs/source/examples/benchmark.rst
index d2ff1cde..b82114ee 100644
--- a/docs/source/examples/benchmark.rst
+++ b/docs/source/examples/benchmark.rst
@@ -19,8 +19,17 @@
Benchmark Usage
================
-Paimon C++ provides Google Benchmark based cases for append-table write/read
and
-primary-key table write/MOR read paths. Benchmarks are disabled by default.
+Paimon C++ provides Google Benchmark based cases at two levels:
+
+``paimon-read-write-benchmark``
+ Table-level cases for append-table write/read and primary-key table
write/MOR
+ read paths.
+
+``paimon-parquet-format-benchmark``
+ Format-level cases that drive the Parquet writer and reader directly,
without
+ catalog lookup, split planning, merge/sort or commit.
+
+Benchmarks are disabled by default.
Build
=====
@@ -29,13 +38,14 @@ Enable benchmarks when configuring CMake::
cmake -S . -B build -DPAIMON_BUILD_BENCHMARKS=ON
cmake --build build --target paimon-read-write-benchmark
+ cmake --build build --target paimon-parquet-format-benchmark
Run all benchmark cases through CTest::
cmake --build build --target benchmark
-Custom Options
-==============
+Table-level Custom Options
+==========================
``paimon-read-write-benchmark`` accepts Google Benchmark options plus the
Paimon
specific options below:
@@ -88,3 +98,97 @@ MOR read from an existing table::
--paimon_source_table_path /path/table \
--paimon_pk_columns=id \
--benchmark_filter=BM_MOR_Read/4
+
+Parquet Format Benchmark
+========================
+
+``paimon-parquet-format-benchmark`` takes only Google Benchmark options. It
+generates its own data and writes it to a temporary directory, so it needs no
+source file or table.
+
+Two things shape how the results should be read:
+
+- Every axis is swept on its own rather than as a combined matrix, so each
+ case answers one question and a change can be attributed to it.
+- Writes go through the local FileSystem into a temporary directory, so
+ absolute numbers carry the cost of that path. Comparisons are meaningful
+ only under the same environment and methodology - the same machine, build
+ configuration and options - which is what makes a before/after comparison
+ useful.
+
+Writer cases (``BM_ParquetWrite_*``) cover flat primitives, ``VARCHAR`` at low
/
+medium / high cardinality with and without file-level dictionary encoding,
+already dictionary-encoded ``VARCHAR`` / ``INTEGER`` input arrays against their
+flat equivalents, ``DECIMAL`` at precision 9 / 18 / 38, nested ``STRUCT`` /
+``LIST`` / ``VECTOR`` / ``MAP``, null density from 0 to 100 percent, rows per
+``AddBatch`` call, column count at a fixed row count, row group size, the
+writer memory threshold that triggers a byte-based row-group flush, and the
+codecs Parquet accepts - ``none``, ``snappy``, ``gzip``, ``brotli``, ``zstd``,
+``lz4_raw`` and ``lz4_hadoop``. Note that ``lz4`` is deliberately not among
+them: it resolves to Arrow's ``LZ4_FRAME``, which
+``parquet::IsCodecSupported`` rejects.
+
+The two dictionary axes are different questions. ``BM_ParquetWrite_String`` and
+``BM_ParquetWrite_StringNoDictionary`` vary whether the *file* is dictionary
+encoded; ``BM_ParquetWrite_Dictionary*`` vary whether the *input array* already
+is, which is what decides whether Arrow can pass indices through to Parquet or
+has to materialize them first.
+
+Reader cases (``BM_ParquetRead_*``) cover full scan, single-column projection,
+predicate-filtered reads at varying selectivity with page-index filtering on
and
+off, skip-heavy reads driven by a strided selection bitmap, null density,
+``DECIMAL`` at precision 9 / 18 / 38, ``DOUBLE``, dictionary-encoded against
+plain-encoded files, rows per ``NextBatch`` call, and nested column reads.
+
+Every case reports ``ns_per_row`` next to ``bytes_per_row`` - ``file_bytes``
for
+writes, ``read_bytes`` for reads - so a change that trades CPU for size is
+visible in both directions. Read cases additionally report ``rows_read``,
+``batches``, and ``row_groups`` / ``row_groups_after_filter`` from the reader's
+own metrics.
+
+Compare filtered cases on ``ns_per_input_row`` and ``bytes_per_input_row``, not
+``ns_per_row`` and ``bytes_per_row``. The latter pair divides by the rows a
case
+actually materialized, so pruning shrinks numerator and denominator together
and
+they can rise even as the run gets faster; the ``_input_row`` pair divides by
the
+rows the file holds, which every setting shares.
+
+``row_groups_after_filter`` counts row groups only. It does not show page-level
+pruning: on the ordered ``id`` column both page-index settings usually keep the
+same row groups, and the page-index gain shows up in ``rows_read``,
+``read_bytes`` and ``ns_per_input_row`` instead.
+
+A case that cannot run - an unsupported codec, a schema the reader rejects -
+calls ``SkipWithError`` and makes the process exit non-zero, so ``ctest -L
+benchmark`` fails instead of reporting a silent skip. Read cases also assert on
+the number of rows they materialized, so a fixture that stopped producing rows
+fails rather than looking fast.
+
+Because the benchmark is only compiled under ``PAIMON_BUILD_BENCHMARKS``, the
+format-layer assumptions it relies on are covered separately by
+``paimon-parquet-format-benchmark-test``, which builds with the normal test
+suite.
+
+Each read case scans a file that is generated once on first use and reused for
+the rest of the run, so a filtered run only pays to build the fixtures its own
+cases need.
+
+All Parquet writer cases::
+
+ paimon-parquet-format-benchmark --benchmark_filter=BM_ParquetWrite
+
+Page-index filtering at 1% selectivity, on and off - compare ``rows_read``,
+``read_bytes`` and ``ns_per_input_row`` between the two::
+
+ paimon-parquet-format-benchmark \
+ --benchmark_filter='BM_ParquetRead_Filtered/keep_pct:1/'
+
+Read batch size sweep, repeated for a stable comparison::
+
+ paimon-parquet-format-benchmark \
+ --benchmark_filter=BM_ParquetRead_BatchSize \
+ --benchmark_repetitions=5 \
+ --benchmark_report_aggregates_only=true
+
+Null density on both sides, to see what definition levels cost::
+
+ paimon-parquet-format-benchmark
--benchmark_filter='Parquet(Write|Read)_Nulls'
