/*
 * 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 "fmt/format.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;
// Only StringFixture lowers this from arrow's 1MB default: at 1MB a 25K-row row group of distinct
// `value_<n>` entries still fits, so no cardinality this file writes would ever make the writer
// fall back to plain and the passthrough gate would have nothing to decline.
constexpr int64_t kDictionaryPageSizeBytes = 64 * 1024;
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)});
}

// One STRING column, so a dictionary case measures one encoder and nothing else.
std::shared_ptr<arrow::Schema> StringSchema() {
    return arrow::schema({MakeField("name", arrow::utf8(), 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. Every batch is built over
// the same alphabet, the shape a single input file produces.
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);
}

// The same, except every batch carries its own dictionary - the shape a compaction rewrite
// produces, where each input file supplies one. A Parquet column chunk holds a single dictionary,
// so arrow keeps the first and falls back to plain encoding for the rest of the row group
// (column_writer.cc, `dictionary->Equals(*preserved_dictionary_)`).
//
// The decoded column is byte for byte what MakeDictionaryStringColumn produces: same values, same
// order, same widths, same cardinality. Only the dictionary object differs, so the delta between
// the two benchmarks is the cost of the fallback and nothing else. Generating a fresh alphabet per
// batch instead would change the data as well - different strings, different widths, a different
// global cardinality - and the comparison would measure all of that at once.
Result<std::shared_ptr<arrow::Array>> MakeChangingDictionaryStringColumn(int64_t num_rows,
                                                                         int64_t offset,
                                                                         int64_t cardinality) {
    // The alphabet is rotated by one position per batch and the indices are shifted the other way,
    // which cancels: index `(offset + cardinality - rotation + i) % cardinality` into an alphabet
    // whose entry `j` is `value_<(j + rotation) % cardinality>` is `value_<(offset + i) %
    // cardinality>` either way. `+ cardinality` keeps the shifted offset non-negative.
    const int64_t rotation = num_rows > 0 ? (offset / num_rows) % cardinality : 0;
    PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Array> values,
                           MakeStringColumn(cardinality, rotation, cardinality));
    return MakeDictionaryColumn(values, num_rows, offset + cardinality - rotation);
}

// 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});
    };
}

// The same, but the batch is typed by the column rather than by the schema, so it can carry an
// encoding the schema does not declare. That is the shape a compaction rewrite produces: the file
// writer is built from the table's logical schema while the reader forwards whatever encoding the
// input file already had, leaving the writer to recover it from the batch.
BatchFactory SingleEncodedColumnBatch(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->field(0)->WithType(column->type())}, {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);
}

// A one-column STRING file at a chosen cardinality, written with a reduced dictionary page limit
// so both regimes the passthrough gate distinguishes are reachable within a 100K-row file. Under
// the limit every data page stays dictionary-encoded and the gate lets the column through; over
// it the writer emits the dictionary page it has and encodes the rest as plain, which is the case
// the gate has to decline. At kDictionaryPageSizeBytes a row group holds roughly 4K distinct
// `value_<n>` entries before overflowing, so cardinality alone picks the regime.
const ReadFixture& StringFixture(int64_t cardinality) {
    const std::string key = fmt::format("string_{}", cardinality);
    return GetFixture(key, [key, cardinality] {
        std::map<std::string, std::string> options;
        options[paimon::parquet::PARQUET_DICTIONARY_PAGE_SIZE] =
            std::to_string(kDictionaryPageSizeBytes);
        return std::make_unique<ReadFixture>(
            key + ".parquet", StringSchema(),
            SingleColumnBatch([cardinality](int64_t rows, int64_t offset) {
                return MakeStringColumn(rows, offset, cardinality);
            }),
            options);
    });
}

// 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=*/{});
}

// arg: dictionary cardinality. The shape the append compaction rewrite actually produces, and the
// one BM_ParquetWrite_DictionaryString does not cover: there the schema itself is a DictionaryType,
// here the writer is built from a plain STRING schema - as a rewrite builds it, from the table's
// logical schema - and the batch arrives dictionary-encoded anyway. The delta against
// BM_ParquetWrite_String at the same cardinality is what the passthrough buys on the write side,
// including the per-batch schema fixup that recovers the encoding from the batch layout.
//
// One alphabet for the whole file, so the writer keeps writing indices: the favourable half of the
// passthrough. BM_ParquetWrite_ChangingDictionaryStringIntoStringSchema is the other half, and a
// rewrite of several input files lands between the two.
void BM_ParquetWrite_DictionaryStringIntoStringSchema(::benchmark::State& state) {
    const int64_t cardinality = state.range(0);
    RunWriteBenchmark(state, StringSchema(),
                      SingleEncodedColumnBatch([cardinality](int64_t rows, int64_t offset) {
                          return MakeDictionaryStringColumn(rows, offset, cardinality);
                      }),
                      kRowsPerBatch, /*options=*/{}, kDefaultCompression);
}

// arg: dictionary cardinality. The same write, except every batch brings its own dictionary, as
// the input files of a compaction do, which is the cost of forwarding an encoding the writer
// cannot reuse. All three of BM_ParquetWrite_String,
// BM_ParquetWrite_DictionaryStringIntoStringSchema and this one write the same logical column at
// the same cardinality, so the triple reads directly: the middle one against the first is what the
// passthrough buys, this one against the middle is what the fallback costs in time, and this one
// against the first is the output size a rewrite that materialized and rebuilt would have produced.
// Row groups here are cut the way they are in production - by size and by the writer's memory
// limit, not at batch boundaries.
void BM_ParquetWrite_ChangingDictionaryStringIntoStringSchema(::benchmark::State& state) {
    const int64_t cardinality = state.range(0);
    RunWriteBenchmark(state, StringSchema(),
                      SingleEncodedColumnBatch([cardinality](int64_t rows, int64_t offset) {
                          return MakeChangingDictionaryStringColumn(rows, offset, cardinality);
                      }),
                      kRowsPerBatch, /*options=*/{}, kDefaultCompression);
}

// 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);
}

// args: string cardinality, and whether the parquet dictionary passthrough is on. With it on, a
// column the file stores dictionary-encoded end to end is handed back as a DictionaryArray instead
// of one materialized value per row, so the pair at a fixed cardinality is what the read half of
// the compaction rewrite saves. At a cardinality high enough that the writer fell back to plain,
// the gate declines and the two runs measure the same work - a divergence there means the gate
// stopped looking at the data page encodings and started trusting the dictionary page.
void BM_ParquetRead_DictionaryPassthrough(::benchmark::State& state) {
    const int64_t cardinality = state.range(0);
    const bool enable_passthrough = state.range(1) != 0;
    std::map<std::string, std::string> options;
    options[paimon::parquet::PARQUET_READ_ENABLE_DICTIONARY_PASSTHROUGH] =
        enable_passthrough ? "true" : "false";
    RunReadBenchmark(state, StringFixture(cardinality), StringSchema(), /*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();
// Same cardinality axis as BM_ParquetWrite_String and BM_ParquetWrite_StringNoDictionary, which
// are its baselines: the three have to line up point for point or the low/medium/high comparison
// cannot be made.
BENCHMARK(BM_ParquetWrite_DictionaryStringIntoStringSchema)
    ->ArgName("cardinality")
    ->Arg(10)
    ->Arg(1000)
    ->Arg(kRowsPerFile)
    ->Unit(benchmark::kMillisecond)
    ->UseRealTime();
// Same axis again, so the three points can be read against the run above them.
BENCHMARK(BM_ParquetWrite_ChangingDictionaryStringIntoStringSchema)
    ->ArgName("cardinality")
    ->Arg(10)
    ->Arg(1000)
    ->Arg(kRowsPerFile)
    ->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();
// The same cardinality axis the write cases use, so the read and write halves of a rewrite can be
// added up at each point. At kRowsPerFile every value is distinct, which overflows
// kDictionaryPageSizeBytes and is the point where the gate has to decline.
BENCHMARK(BM_ParquetRead_DictionaryPassthrough)
    ->ArgNames({"cardinality", "passthrough"})
    ->Args({10, 0})
    ->Args({10, 1})
    ->Args({1000, 0})
    ->Args({1000, 1})
    ->Args({kRowsPerFile, 0})
    ->Args({kRowsPerFile, 1})
    ->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;
}