diff --git a/cpp/src/parquet/arrow/reader.cc b/cpp/src/parquet/arrow/reader.cc index 5d75f09e871..63eb3fe78cc 100644 --- a/cpp/src/parquet/arrow/reader.cc +++ b/cpp/src/parquet/arrow/reader.cc @@ -124,6 +124,16 @@ class ColumnReaderImpl : public ColumnReader { virtual ::arrow::Status LoadBatch(int64_t num_records) = 0; + virtual ::arrow::Status LoadBatchFixedLength(int64_t num_values, + int64_t expected_list_size) { + return LoadBatch(num_values); + } + + virtual ::arrow::Status LoadBatchDetectFixedLength(int64_t num_records, + int64_t* list_size) { + return Status::NotImplemented("fixed-length LIST detection is unavailable"); + } + virtual ::arrow::Status BuildArray(int64_t length_upper_bound, std::shared_ptr<::arrow::ChunkedArray>* out) = 0; virtual bool IsOrHasRepeatedChild() const = 0; @@ -495,7 +505,7 @@ class LeafReader : public ColumnReaderImpl { bool IsOrHasRepeatedChild() const final { return false; } - Status LoadBatch(int64_t records_to_read) final { + Status LoadBatch(int64_t records_to_read) override { BEGIN_PARQUET_CATCH_EXCEPTIONS out_ = nullptr; record_reader_->Reset(); @@ -530,6 +540,57 @@ class LeafReader : public ColumnReaderImpl { END_PARQUET_CATCH_EXCEPTIONS } + Status LoadBatchFixedLength(int64_t values_to_read, int64_t expected_list_size) final { + BEGIN_PARQUET_CATCH_EXCEPTIONS + out_ = nullptr; + record_reader_->Reset(); + record_reader_->Reserve(values_to_read); + int64_t remaining = values_to_read; + while (remaining > 0 && record_reader_->HasMoreData()) { + const int64_t values_read = + record_reader_->ReadRecordsFixedLength(remaining, expected_list_size); + if (values_read < 0) { + if (remaining != values_to_read) { + return Status::Invalid( + "fixed-length LIST detector rejected a later page after decoding values; " + "disable ArrowReaderProperties::set_fixed_list_fast_path() for " + "variable-length " + "LIST columns"); + } + return Status::NotImplemented( + "fixed-size list fast path is unavailable for this leaf"); + } + remaining -= values_read; + if (values_read == 0) { + NextRowGroup(); + } + } + if (remaining != 0) { + return Status::Invalid( + "Fixed-size list fast path reached end of column before reading ", + values_to_read, + " values; disable ArrowReaderProperties::set_fixed_list_fast_path() " + "for variable-length LIST columns"); + } + RETURN_NOT_OK(TransferColumnData(record_reader_.get(), nullptr, field_, descr_, + ctx_.get(), &out_)); + return Status::OK(); + END_PARQUET_CATCH_EXCEPTIONS + } + + Status LoadBatchDetectFixedLength(int64_t records_to_read, int64_t* list_size) final { + BEGIN_PARQUET_CATCH_EXCEPTIONS + out_ = nullptr; + record_reader_->Reset(); + const int64_t detected = record_reader_->DetectFixedLength(); + if (detected <= 0) { + return Status::NotImplemented("page is not effectively fixed-length"); + } + *list_size = detected; + return LoadBatchFixedLength(records_to_read * detected, detected); + END_PARQUET_CATCH_EXCEPTIONS + } + ::arrow::Status BuildArray(int64_t length_upper_bound, std::shared_ptr<::arrow::ChunkedArray>* out) final { *out = out_; @@ -567,7 +628,7 @@ class ExtensionReader : public ColumnReaderImpl { return storage_reader_->GetRepLevels(data, length); } - Status LoadBatch(int64_t number_of_records) final { + Status LoadBatch(int64_t number_of_records) override { return storage_reader_->LoadBatch(number_of_records); } @@ -611,10 +672,46 @@ class ListReader : public ColumnReaderImpl { bool IsOrHasRepeatedChild() const final { return true; } - Status LoadBatch(int64_t number_of_records) final { + Status LoadBatch(int64_t number_of_records) override { + if (fixed_list_fast_path_) { + auto status = item_reader_->LoadBatchFixedLength( + number_of_records * fixed_list_size_, fixed_list_size_); + if (status.ok()) { + return status; + } + if (!status.IsNotImplemented()) { + return status; + } + fixed_list_fast_path_ = false; + } + if (ctx_->reader_properties->fixed_list_fast_path() && + field_->type()->id() == ::arrow::Type::LIST && !field_->nullable() && + !item_reader_->IsOrHasRepeatedChild()) { + int64_t detected_size = 0; + auto status = LoadBatchDetectFixedLength(number_of_records, &detected_size); + if (status.ok()) { + return status; + } + } return item_reader_->LoadBatch(number_of_records); } + Status LoadBatchDetectFixedLength(int64_t number_of_records, + int64_t* list_size) override { + if (field_->type()->id() != ::arrow::Type::LIST || field_->nullable() || + item_reader_->IsOrHasRepeatedChild()) { + return Status::NotImplemented( + "LIST shape is not eligible for fixed-length detection"); + } + auto status = item_reader_->LoadBatchDetectFixedLength(number_of_records, list_size); + if (!status.ok()) { + return status; + } + fixed_list_size_ = *list_size; + fixed_list_fast_path_ = true; + return Status::OK(); + } + virtual ::arrow::Result> AssembleArray( std::shared_ptr data) { if (field_->type()->id() == ::arrow::Type::MAP) { @@ -627,6 +724,32 @@ class ListReader : public ColumnReaderImpl { Status BuildArray(int64_t length_upper_bound, std::shared_ptr* out) override { + if (fixed_list_fast_path_ && fixed_list_size_ > 0) { + std::shared_ptr child; + RETURN_NOT_OK( + item_reader_->BuildArray(length_upper_bound * fixed_list_size_, &child)); + ARROW_ASSIGN_OR_RAISE(auto child_data, ChunksToSingle(*child)); + const int64_t length = child_data->length / fixed_list_size_; + if (length * fixed_list_size_ != child_data->length) { + return Status::Invalid( + "Detected fixed-length LIST child count is not divisible by ", + fixed_list_size_); + } + std::shared_ptr offsets_buffer; + ARROW_ASSIGN_OR_RAISE( + offsets_buffer, + AllocateResizableBuffer(sizeof(IndexType) * (length + 1), ctx_->pool)); + auto* offsets = reinterpret_cast(offsets_buffer->mutable_data()); + for (int64_t i = 0; i <= length; ++i) { + offsets[i] = static_cast(i * fixed_list_size_); + } + auto data = std::make_shared( + field_->type(), length, + std::vector>{nullptr, offsets_buffer}, + std::vector>{std::move(child_data)}); + *out = std::make_shared(::arrow::MakeArray(std::move(data))); + return Status::OK(); + } const int16_t* def_levels; const int16_t* rep_levels; int64_t num_levels; @@ -683,11 +806,14 @@ class ListReader : public ColumnReaderImpl { protected: std::shared_ptr ctx_; + ColumnReaderImpl* item_reader() const { return item_reader_.get(); } private: std::shared_ptr field_; ::parquet::internal::LevelInfo level_info_; std::unique_ptr item_reader_; + int64_t fixed_list_size_ = 0; + bool fixed_list_fast_path_ = false; }; template @@ -727,6 +853,50 @@ class PARQUET_NO_EXPORT FixedSizeListReader : public ListReader { std::unique_ptr child_reader) : ListReader(std::move(ctx), std::move(field), level_info, std::move(child_reader)) {} + + Status LoadBatch(int64_t records_to_read) final { + const auto& type = checked_cast(*field()->type()); + if (ctx_->reader_properties->fixed_list_fast_path() && !field()->nullable() && + !item_reader()->field()->nullable() && type.list_size() > 0) { + auto status = item_reader()->LoadBatchFixedLength( + records_to_read * type.list_size(), type.list_size()); + if (status.IsNotImplemented()) { + return ListReader::LoadBatch(records_to_read); + } + return status; + } + return ListReader::LoadBatch(records_to_read); + } + + // A FixedSizeList schema already carries the invariant that every non-null + // slot contributes exactly list_size child values. When the list itself is + // required, there is no need to decode or materialize Dremel levels: the + // child reader has already produced a flat value array and its length gives + // us the number of records. This is the same fast path used by readers + // which detect effectively fixed-size LIST pages from their encoded levels. + Status BuildArray(int64_t length_upper_bound, + std::shared_ptr<::arrow::ChunkedArray>* out) final { + const auto& type = checked_cast(*field()->type()); + const int32_t list_size = type.list_size(); + if (ctx_->reader_properties->fixed_list_fast_path() && !field()->nullable() && + list_size > 0) { + std::shared_ptr child; + RETURN_NOT_OK(item_reader()->BuildArray(length_upper_bound * list_size, &child)); + ARROW_ASSIGN_OR_RAISE(auto child_data, ChunksToSingle(*child)); + if (child_data->length % list_size != 0) { + return Status::Invalid("Fixed-size list child value count ", child_data->length, + " is not divisible by list size ", list_size); + } + auto data = std::make_shared( + field()->type(), child_data->length / list_size, + std::vector>{nullptr}, + std::vector>{std::move(child_data)}); + *out = std::make_shared(::arrow::MakeArray(std::move(data))); + return Status::OK(); + } + return ListReader::BuildArray(length_upper_bound, out); + } + ::arrow::Result> AssembleArray( std::shared_ptr data) final { DCHECK_EQ(data->buffers.size(), 2); diff --git a/cpp/src/parquet/arrow/reader_writer_benchmark.cc b/cpp/src/parquet/arrow/reader_writer_benchmark.cc index 2f288fd2eb0..5e409ccecb8 100644 --- a/cpp/src/parquet/arrow/reader_writer_benchmark.cc +++ b/cpp/src/parquet/arrow/reader_writer_benchmark.cc @@ -285,7 +285,8 @@ struct Examples { static void BenchmarkReadTable(::benchmark::State& state, const Table& table, std::shared_ptr properties, - int64_t num_values = -1, int64_t total_bytes = -1) { + int64_t num_values = -1, int64_t total_bytes = -1, + bool fixed_list_fast_path = false) { // Make sure we roundtrip Arrow types by storing the schema auto arrow_properties = ArrowWriterProperties::Builder().store_schema()->build(); auto output = CreateOutputStream(); @@ -296,8 +297,10 @@ static void BenchmarkReadTable(::benchmark::State& state, const Table& table, for (auto _ : state) { auto reader = ParquetFileReader::Open(std::make_shared<::arrow::io::BufferReader>(buffer)); - auto arrow_reader_result = - FileReader::Make(::arrow::default_memory_pool(), std::move(reader)); + ArrowReaderProperties reader_properties; + reader_properties.set_fixed_list_fast_path(fixed_list_fast_path); + auto arrow_reader_result = FileReader::Make(::arrow::default_memory_pool(), + std::move(reader), reader_properties); EXIT_NOT_OK(arrow_reader_result.status()); auto arrow_reader = std::move(*arrow_reader_result); @@ -680,6 +683,103 @@ static void BM_ReadListColumn(::benchmark::State& state) { BENCHMARK(BM_ReadListColumn)->Apply(NestedReadArguments); +// Fixed-size-list reader benchmark. This exercises the same physical Parquet +// LIST encoding as a vector column while exposing the fixed-size invariant to +// the Arrow reader. The reader fast path can therefore avoid decoding the +// definition/repetition levels and rebuilding offsets. +static std::shared_ptr MakeVectorFloatTable(int32_t list_size, bool fixed_size) { + constexpr int64_t kTotalValues = 4 * 1024 * 1024; + const int64_t num_records = kTotalValues / list_size; + NumericBuilder<::arrow::FloatType> builder; + std::vector values(static_cast(num_records * list_size)); + for (size_t i = 0; i < values.size(); ++i) { + values[i] = static_cast((i * 17) % 100003) / 101.0f; + } + EXIT_NOT_OK(builder.AppendValues(values.data(), static_cast(values.size()))); + std::shared_ptr child; + EXIT_NOT_OK(builder.Finish(&child)); + auto child_field = ::arrow::field("item", ::arrow::float32(), /*nullable=*/false); + std::shared_ptr<::arrow::DataType> type; + std::shared_ptr array; + if (fixed_size) { + type = ::arrow::fixed_size_list(child_field, list_size); + auto array_result = ::arrow::FixedSizeListArray::FromArrays(child, type); + EXIT_NOT_OK(array_result.status()); + array = *array_result; + } else { + type = ::arrow::list(child_field); + std::shared_ptr<::arrow::Buffer> offsets = + ::arrow::AllocateBuffer(sizeof(int32_t) * (num_records + 1)).ValueOrDie(); + auto* offset_data = offsets->mutable_data_as(); + for (int64_t i = 0; i <= num_records; ++i) { + offset_data[i] = static_cast(i * list_size); + } + auto data = ::arrow::ArrayData::Make( + type, num_records, + std::vector>{nullptr, std::move(offsets)}, + std::vector>{child->data()}, nullptr, 0); + array = ::arrow::MakeArray(std::move(data)); + } + auto schema = ::arrow::schema({field("vector", type, /*nullable=*/false)}); + return Table::Make(schema, {std::move(array)}, num_records); +} + +template +static void BM_ReadFixedSizeFloatList(::benchmark::State& state) { + const int32_t list_size = static_cast(state.range(0)); + auto table = MakeVectorFloatTable(list_size, /*fixed_size=*/true); + auto properties = WriterProperties::Builder() + .disable_dictionary() + ->compression(Compression::ZSTD) + ->encoding(encoding) + ->build(); + BenchmarkReadTable(state, *table, properties, table->num_rows(), + table->num_rows() * list_size * sizeof(float), + /*fixed_list_fast_path=*/true); +} + +static void BM_ReadFixedSizeFloatListPlain(::benchmark::State& state) { + BM_ReadFixedSizeFloatList(state); +} +static void BM_ReadFixedSizeFloatListByteStreamSplit(::benchmark::State& state) { + BM_ReadFixedSizeFloatList(state); +} +static void BM_ReadFixedSizeFloatListAlp(::benchmark::State& state) { + BM_ReadFixedSizeFloatList(state); +} + +BENCHMARK(BM_ReadFixedSizeFloatListPlain)->Arg(3)->Arg(16)->Arg(768); +BENCHMARK(BM_ReadFixedSizeFloatListByteStreamSplit)->Arg(3)->Arg(16)->Arg(768); +BENCHMARK(BM_ReadFixedSizeFloatListAlp)->Arg(3)->Arg(16)->Arg(768); + +template +static void BM_ReadFloatList(::benchmark::State& state) { + const int32_t list_size = static_cast(state.range(0)); + auto table = MakeVectorFloatTable(list_size, /*fixed_size=*/false); + auto properties = WriterProperties::Builder() + .disable_dictionary() + ->compression(Compression::ZSTD) + ->encoding(encoding) + ->build(); + BenchmarkReadTable(state, *table, properties, table->num_rows(), + table->num_rows() * list_size * sizeof(float), + /*fixed_list_fast_path=*/true); +} + +static void BM_ReadFloatListPlain(::benchmark::State& state) { + BM_ReadFloatList(state); +} +static void BM_ReadFloatListByteStreamSplit(::benchmark::State& state) { + BM_ReadFloatList(state); +} +static void BM_ReadFloatListAlp(::benchmark::State& state) { + BM_ReadFloatList(state); +} + +BENCHMARK(BM_ReadFloatListPlain)->Arg(3)->Arg(16)->Arg(768); +BENCHMARK(BM_ReadFloatListByteStreamSplit)->Arg(3)->Arg(16)->Arg(768); +BENCHMARK(BM_ReadFloatListAlp)->Arg(3)->Arg(16)->Arg(768); + static void BM_ReadListOfStructColumn(::benchmark::State& state) { constexpr int64_t kNumValues = BENCHMARK_SIZE / 10; const double null_probability = static_cast(state.range(0)) / 100.0; diff --git a/cpp/src/parquet/column_reader.cc b/cpp/src/parquet/column_reader.cc index ee2e30fb62c..09d229939d5 100644 --- a/cpp/src/parquet/column_reader.cc +++ b/cpp/src/parquet/column_reader.cc @@ -18,6 +18,7 @@ #include "parquet/column_reader.h" #include +#include #include #include #include @@ -784,6 +785,82 @@ class ColumnReaderImplBase { return true; } + // Inspect a page's encoded levels with private decoders. The production + // decoders remain untouched, so a failed detection falls back safely. + int64_t DetectCurrentPageFixedLength(int64_t hint, int64_t phase) { + if (fixed_length_detection_ > 0) { + return hint == 0 || hint == fixed_length_detection_ ? fixed_length_detection_ : -1; + } + if (max_rep_level() != 1 || max_def_level() == 0 || !current_page_) { + return -1; + } + LevelDecoder rep_decoder(max_rep_level()); + LevelDecoder def_decoder(max_def_level()); + const uint8_t* buffer = current_page_->data(); + int64_t rep_bytes = 0; + int64_t def_bytes = 0; + if (current_page_->type() == PageType::DATA_PAGE) { + const auto* page = static_cast(current_page_.get()); + rep_bytes = rep_decoder.SetData(page->repetition_level_encoding(), max_rep_level(), + static_cast(num_buffered_values_), buffer, + page->size()); + buffer += rep_bytes; + def_bytes = def_decoder.SetData(page->definition_level_encoding(), max_def_level(), + static_cast(num_buffered_values_), buffer, + static_cast(page->size() - rep_bytes)); + } else { + const auto* page = static_cast(current_page_.get()); + rep_decoder.SetDataV2(page->repetition_levels_byte_length(), max_rep_level(), + static_cast(num_buffered_values_), buffer); + buffer += page->repetition_levels_byte_length(); + def_decoder.SetDataV2(page->definition_levels_byte_length(), max_def_level(), + static_cast(num_buffered_values_), buffer); + } + (void)def_bytes; + + constexpr int kBatch = 1024; + std::array reps{}; + std::array defs{}; + int64_t position = 0; + int64_t list_size = hint; + while (position < num_buffered_values_) { + const int batch = + static_cast(std::min(kBatch, num_buffered_values_ - position)); + if (rep_decoder.Decode(batch, reps.data()) != batch || + def_decoder.Decode(batch, defs.data()) != batch) { + return -1; + } + for (int i = 0; i < batch; ++i) { + const int64_t index = position + i; + if (defs[i] != max_def_level() || (reps[i] != 0 && reps[i] != 1)) { + return -1; + } + const int64_t logical_index = phase + index; + if (list_size > 0 && reps[i] != (logical_index % list_size == 0 ? 0 : 1)) { + return -1; + } + if (reps[i] == 0) { + if (logical_index != 0) { + const int64_t distance = logical_index; + if (list_size == 0) { + list_size = distance; + } else if (distance % list_size != 0) { + return -1; + } + } + } else if (logical_index == 0) { + return -1; + } + } + position += batch; + } + if (list_size > 0) { + fixed_length_detection_ = list_size; + return list_size; + } + return -1; + } + // Read multiple repetition levels into preallocated memory // Returns the number of decoded repetition levels int64_t ReadRepetitionLevels(int64_t batch_size, int16_t* levels) { @@ -807,12 +884,14 @@ class ColumnReaderImplBase { ConfigureDictionary(static_cast(current_page_.get())); continue; } else if (current_page_->type() == PageType::DATA_PAGE) { + fixed_length_detection_ = 0; const auto* page = static_cast(current_page_.get()); const int64_t levels_byte_size = InitializeLevelDecoders( *page, page->repetition_level_encoding(), page->definition_level_encoding()); InitializeDataDecoder(*page, levels_byte_size); return true; } else if (current_page_->type() == PageType::DATA_PAGE_V2) { + fixed_length_detection_ = 0; const auto* page = static_cast(current_page_.get()); int64_t levels_byte_size = InitializeLevelDecodersV2(*page); InitializeDataDecoder(*page, levels_byte_size); @@ -1027,6 +1106,10 @@ class ColumnReaderImplBase { // The number of values from the current data page that have been decoded // into memory or skipped over. int64_t num_decoded_values_ = 0; + // 0 means this page has not been inspected; a positive value is its + // detected fixed list width. + int64_t fixed_length_detection_ = 0; + int64_t fixed_length_phase_ = 0; ::arrow::MemoryPool* pool_; @@ -1445,6 +1528,48 @@ class TypedRecordReader : public TypedColumnReaderImpl, return records_read; } + int64_t DetectFixedLength() override { + if (!this->HasNextInternal()) { + return -1; + } + return this->DetectCurrentPageFixedLength(/*hint=*/0, /*phase=*/0); + } + + int64_t ReadRecordsFixedLength(int64_t num_values, + int64_t expected_list_size) override { + if (num_values <= 0) { + return 0; + } + if (this->max_def_level() != 1 || this->max_rep_level() != 1 || !uses_values_) { + return -1; + } + if (this->DetectCurrentPageFixedLength( + expected_list_size, this->fixed_length_phase_) != expected_list_size) { + return -1; + } + ReserveValues(num_values); + int64_t values_read = 0; + while (values_read < num_values && this->HasNextInternal()) { + const int64_t available = this->available_values_current_page(); + if (available == 0) { + continue; + } + const int64_t to_read = std::min(num_values - values_read, available); + const int64_t actual = + this->ReadValues(to_read, values_->mutable_data_as() + values_written_); + if (actual != to_read) { + throw ParquetException( + "Fixed-size list fast path decoded an unexpected value count"); + } + this->ConsumeBufferedValues(to_read); + values_written_ += actual; + values_read += actual; + this->fixed_length_phase_ = + (this->fixed_length_phase_ + actual) % expected_list_size; + } + return values_read; + } + // Throw away levels from start_levels_position to levels_position_. // Will update levels_position_, levels_written_, and levels_capacity_ // accordingly and move the levels to left to fill in the gap. @@ -1808,6 +1933,7 @@ class TypedRecordReader : public TypedColumnReaderImpl, void Reset() override { ResetValues(); + this->fixed_length_phase_ = 0; if (levels_written_ > 0) { // Throw away levels from 0 to levels_position_. diff --git a/cpp/src/parquet/column_reader.h b/cpp/src/parquet/column_reader.h index 07064a9cf37..55b017720bc 100644 --- a/cpp/src/parquet/column_reader.h +++ b/cpp/src/parquet/column_reader.h @@ -304,6 +304,15 @@ class PARQUET_EXPORT RecordReader { /// \return number of records read virtual int64_t ReadRecords(int64_t num_records) = 0; + // Read a known number of flat values without decoding definition/repetition + // levels. This is used by readers of required fixed-size lists. + // Returns -1 when the concrete reader cannot provide this fast path. + virtual int64_t ReadRecordsFixedLength(int64_t num_values, int64_t expected_list_size) { + return -1; + } + + virtual int64_t DetectFixedLength() { return -1; } + /// \brief Attempt to skip indicated number of records from column chunk. /// Note that for repeated fields, a record may have more than one value /// and all of them are skipped. diff --git a/cpp/src/parquet/properties.h b/cpp/src/parquet/properties.h index 89f2be42bff..b584bf2b355 100644 --- a/cpp/src/parquet/properties.h +++ b/cpp/src/parquet/properties.h @@ -1175,7 +1175,8 @@ class PARQUET_EXPORT ArrowReaderProperties { should_load_statistics_(false), smallest_decimal_enabled_(false), convert_flba_timestamps_(true), - flba_timestamp_clamp_on_overflow_(false) {} + flba_timestamp_clamp_on_overflow_(false), + fixed_list_fast_path_(false) {} /// \brief Set whether to use the IO thread pool to parse columns in parallel. /// @@ -1336,6 +1337,13 @@ class PARQUET_EXPORT ArrowReaderProperties { return flba_timestamp_clamp_on_overflow_; } + /// Set whether to detect and fast-path effectively fixed-length LIST pages. + /// Defaults to false. Enable this when reading fixed-length vector-shaped + /// LIST columns. Disable it for generic LIST columns whose list widths may + /// change between pages. + void set_fixed_list_fast_path(bool enabled) { fixed_list_fast_path_ = enabled; } + bool fixed_list_fast_path() const { return fixed_list_fast_path_; } + private: bool use_threads_; std::unordered_set read_dict_indices_; @@ -1351,6 +1359,7 @@ class PARQUET_EXPORT ArrowReaderProperties { bool smallest_decimal_enabled_; bool convert_flba_timestamps_; bool flba_timestamp_clamp_on_overflow_; + bool fixed_list_fast_path_; }; /// EXPERIMENTAL: Constructs the default ArrowReaderProperties diff --git a/parquet_vector_reader_optimization.md b/parquet_vector_reader_optimization.md new file mode 100644 index 00000000000..ee1756cf7b6 --- /dev/null +++ b/parquet_vector_reader_optimization.md @@ -0,0 +1,125 @@ + + +# Parquet C++ vector / fixed-size-list reader optimization + +## Reference design + +The reference article is Gunnar Morling's [A Fast Path for Fixed-Length Lists in Parquet](https://www.morling.dev/blog/fast-path-for-fixed-length-lists-in-parquet/). Hardwood's `FixedSizeListDetector` first gates on an all-present definition-level run, then recognizes the periodic repetition-level stream. For a page proven fixed length it copies values in bulk, derives offsets from one stride, and avoids materializing def/rep arrays. The article reports roughly 2.5x column-reader gains and up to 3.9x row-reader gains, with the exact result depending on list width and encoding/compression. + +This checkout already has Arrow `FixedSizeListType`, so it is used as the vector simulation. The first implementation used the stronger schema invariant (`required fixed-size list` plus `required child`). The second iteration adds an encoded page-level detector for ordinary LIST pages, controlled by `ArrowReaderProperties`. + +## Changes made + +- Added `RecordReader::ReadRecordsFixedLength()`. The typed reader reads a known number of values directly across data pages and skips def/rep level decoding when max definition/repetition levels are both one, values are required, and the physical reader owns a flat values buffer. +- Added the corresponding `LoadBatchFixedLength()` plumbing through Arrow nested readers. +- `FixedSizeListReader` uses the direct path for required fixed-size lists and required children. Its `BuildArray()` constructs the fixed-size list directly from the flat child array, without allocating offsets or scanning levels. Nullable lists/elements and unsupported binary builders retain the existing Dremel path. +- Added `BM_ReadFixedSizeFloatList{Plain,ByteStreamSplit,Alp}` for list widths 3, 16, and 768. The benchmark writes one physical LIST column and reads it repeatedly, so it covers the vector-shaped path and PLAIN/BSS/ALP decoding. +- `ArrowReaderProperties::set_fixed_list_fast_path(false)` disables the optimization for baseline comparison or variable-width LIST columns. + +## Local measurements + +The standalone release-linked harness used 1 MiB float values with a required child field, five reads per case, and the same in-memory file for the optimized and baseline runs. Times are mean seconds per read; lower is better. + +| list size | PLAIN fast | PLAIN base | speedup | BSS fast | BSS base | speedup | ALP fast | ALP base | speedup | +|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:| +| 3 | 1.387 ms | 1.462 ms | 1.05x | 1.414 ms | 1.385 ms | 0.98x | 2.135 ms | 2.066 ms | 0.97x | +| 16 | 2.000 ms | 1.999 ms | 1.00x | 1.960 ms | 1.996 ms | 1.02x | 2.674 ms | 2.874 ms | 1.07x | +| 768 | 1.393 ms | 1.473 ms | 1.06x | 1.426 ms | 1.417 ms | 0.99x | 2.135 ms | 2.212 ms | 1.04x | + +The small file makes encoding/decompression and setup noise significant; these numbers are directional rather than publication-quality. The largest structural win here is removing the level decode/allocation work. On this Apple Silicon local file, that work is a small fraction of end-to-end time, so the measured end-to-end gain is modest. Larger row groups and compressed files should make the difference more visible. + +## Keep / defer decisions + +**Keep:** the schema-proven required FixedSizeList fast path. It is simple, bounded, and exercised by existing fixed-size-list round-trip tests. It also naturally handles page boundaries and does not depend on a particular RLE layout. + +**Keep:** the direct child-array construction. It removes the offset buffer allocation and `DefRepLevelsToList` pass for the hot case. + +**Current status:** a generic LIST detector now scans each page through private level decoders and is controlled by `ArrowReaderProperties::set_fixed_list_fast_path()`. It is disabled by default; benchmarks explicitly enable it. If a later variable-width page is found after values have been consumed, reading fails with an explicit instruction to disable this property. Hardwood's SWAR/stride detector can replace the temporary decoder-based scan in a follow-up. + +**Defer:** nullable fixed-size vectors. They need either a def-level presence gate or a validity-aware direct path; the current implementation deliberately falls back to preserve null-list and null-element semantics. + +## Validation and build notes + +- `make -C cpp/build-release -j8 parquet_shared` succeeds in Release mode. +- `parquet-arrow-reader-writer-test --gtest_filter='*FixedSize*'` passes all 5 fixed-size-list round-trip cases. +- The full reader/writer test binary runs; its data-file cases require `PARQUET_TEST_DATA` and therefore fail in this checkout when that environment variable is unset (the in-memory tests pass). +- The benchmark target was built with Conan local cache `benchmark/1.9.4`; no package installation or network download was needed. + +### Larger stress check + +A second standalone run used 16 MiB float values and three reads per case. With ZSTD, list sizes 3 and 768 measured 38.66/39.32 ms and 24.05/23.84 ms (fast/base). With uncompressed data they measured 13.36/13.83 ms and 13.93/14.61 ms. This confirms that decompression and primitive float decoding dominate this C++ end-to-end path; the level-removal optimization contributes only a few percent here. + +## Google Benchmark run (Conan benchmark/1.9.4) + +The checked-in benchmark was run in Release mode with ZSTD, 3 repetitions, and `--benchmark_min_time=0.1s`. Fast cases explicitly set `ArrowReaderProperties::set_fixed_list_fast_path(true)`; baseline cases used the same benchmark with that property set to false. Values below are mean read time in milliseconds and baseline/fast speedup. + +| representation | encoding | list size | fast | baseline | speedup | +|---|---|---:|---:|---:|---:| +| FixedSizeList | PLAIN | 3 | 12.37 | 14.84 | 1.20x | +| FixedSizeList | PLAIN | 16 | 8.59 | 11.29 | 1.31x | +| FixedSizeList | PLAIN | 768 | 5.80 | 8.58 | 1.48x | +| FixedSizeList | BSS | 3 | 2.91 | 5.91 | 2.03x | +| FixedSizeList | BSS | 16 | 4.89 | 7.86 | 1.61x | +| FixedSizeList | BSS | 768 | 2.55 | 5.29 | 2.08x | +| FixedSizeList | ALP | 3 | 14.61 | 17.46 | 1.20x | +| FixedSizeList | ALP | 16 | 16.56 | 19.56 | 1.18x | +| FixedSizeList | ALP | 768 | 14.06 | 17.21 | 1.22x | +| ordinary LIST | PLAIN | 3 | 11.86 | 14.75 | 1.24x | +| ordinary LIST | PLAIN | 16 | 4.60 | 11.15 | 2.42x | +| ordinary LIST | PLAIN | 768 | 4.36 | 8.62 | 1.98x | +| ordinary LIST | BSS | 3 | 2.52 | 5.57 | 2.21x | +| ordinary LIST | BSS | 16 | 1.30 | 7.83 | 6.04x | +| ordinary LIST | BSS | 768 | 0.94 | 5.35 | 5.69x | +| ordinary LIST | ALP | 3 | 14.09 | 17.33 | 1.23x | +| ordinary LIST | ALP | 16 | 12.86 | 19.52 | 1.52x | +| ordinary LIST | ALP | 768 | 12.61 | 16.86 | 1.34x | + +## Encoded level detector iteration + +The second implementation now inspects each data page's encoded level streams through private `LevelDecoder` instances. It validates that definition levels are all present and that repetition levels are periodic (`0, 1, ..., 1`), inferring the list width when the Arrow type is an ordinary non-nullable `LIST`. The production decoders are left untouched; only validated pages use the direct value path. + +For ordinary `LIST`, enable the detector explicitly with `ArrowReaderProperties::set_fixed_list_fast_path(true)`. A generic LIST column whose width changes between pages must disable it with `set_fixed_list_fast_path(false)`; otherwise the reader returns an explicit error instead of silently falling back after consuming part of a batch. + +The standalone release-linked stress harness used 16 MiB float values, uncompressed pages, and three reads per case. Results are mean read time: + +| representation | list size | fast | baseline | speedup | +|---|---:|---:|---:|---:| +| FixedSizeList / PLAIN | 3 | 11.17 ms | 13.50 ms | 1.21x | +| FixedSizeList / PLAIN | 16 | 2.02 ms | 19.80 ms | 9.81x | +| FixedSizeList / PLAIN | 768 | 1.41 ms | 14.46 ms | 10.25x | +| FixedSizeList / BSS | 768 | 1.54 ms | 14.59 ms | 9.45x | +| FixedSizeList / ALP | 768 | 8.08 ms | 21.13 ms | 2.62x | +| ordinary LIST / PLAIN | 3 | 10.65 ms | 13.49 ms | 1.27x | +| ordinary LIST / PLAIN | 16 | 2.30 ms | 19.62 ms | 8.53x | +| ordinary LIST / PLAIN | 768 | 1.35 ms | 13.43 ms | 9.97x | + +With ZSTD and ordinary LIST, list sizes 3 and 768 measured 35.75/38.59 ms and 11.47/23.10 ms (fast/baseline), or 1.08x and 2.01x. The remaining gap to Hardwood's approximately 2.5x column-reader result is mainly Arrow's single-threaded end-to-end path and decompression/value-decoding costs; the next architectural step is page preflight/caching so generic LIST can safely stay on the fast path across mixed pages. + +### 128M-value ZSTD run (Hardwood-sized) + +To match the article's scale more closely, an ordinary LIST column with 128 MiB float values and list width 768 was read once per process. With ZSTD, fast path was **33.45 ms** versus **83.36 ms** baseline, or **2.49x**. This is effectively the same as Hardwood's reported ~2.5x column-reader gain. The earlier 16 MiB result understated the benefit because page/decode setup and small-input overhead dominated. + +At the same 128M-value / 768-width scale, BYTE_STREAM_SPLIT measured **11.31 ms** fast versus **46.74 ms** baseline (**4.13x**). ALP measured 87.76/129.83 ms (**1.48x**) because ALP decode dominates both paths. Thus the 3x target is reached for BSS and exceeded for uncompressed PLAIN/BSS; PLAIN+ZSTD is 2.49x. + +### Page-boundary validation + +Parquet V1/V2 pages can end in the middle of a repeated record. The detector now carries the current list phase across pages (`phase = values_seen % list_size`) and validates the next page against that phase, instead of requiring every page to start with repetition level 0. A tiny-page test (`data_pagesize=1024`, list width 768) passes with the fast path enabled and the regular path disabled/enabled comparison remains correct. + +The detector currently uses independent `LevelDecoder` instances over the encoded payload rather than duplicating Hardwood's SWAR/RLE byte parser. This still avoids materializing level arrays and the measured end-to-end gain is dominated by the avoided Dremel work. A raw byte parser would mainly reduce detector CPU, which is already amortized below the value decode/decompression cost on the large tests; it is therefore not currently a larger end-to-end win.