Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
67 commits
Select commit Hold shift + click to select a range
6b750f4
Move TypedColReaderImpl::Skip
AntoinePrv Jul 9, 2026
17f1c9a
Break TypedRecordReader inheritance on TypedColumnReader
AntoinePrv Jul 9, 2026
51b0169
Add ReadValuesBuffer
AntoinePrv Jul 10, 2026
a1f7c22
Split ColumnReaderImplBase definition and declaration
AntoinePrv Jul 10, 2026
aa31656
Add DefRepLevelsDecoder
AntoinePrv Jul 10, 2026
737cd10
Explicit int32_t for LevelDecoder
AntoinePrv Jul 10, 2026
4ef6eef
Change ValuesBuffer API
AntoinePrv Jul 15, 2026
93b069a
Add StackedLevelDecoder and prepare for modularity
AntoinePrv Jul 15, 2026
3ce765e
Remove DefRepLevelsDecoder
AntoinePrv Jul 15, 2026
05d6018
Split TypedRecordReader implementation
AntoinePrv Jul 16, 2026
b1baa1c
First RequiredTypedRecordReader
AntoinePrv Jul 16, 2026
9f17b93
Remove RecordReader protected members
AntoinePrv Jul 16, 2026
1edbd7d
Fully use RequiredRecordReader
AntoinePrv Jul 16, 2026
5e79853
Rename ReadValues -> ValueSink
AntoinePrv Jul 17, 2026
5c413aa
Split ReserveValuesAndIsValid
AntoinePrv Jul 17, 2026
7f40e5e
Devirtualize hooks with CRTP
AntoinePrv Jul 20, 2026
218bdc1
Simplify template
AntoinePrv Jul 20, 2026
6f718c0
Remove StackedLevelDecoder
AntoinePrv Jul 20, 2026
9eeb3a6
Remove mark_as_written
AntoinePrv Jul 20, 2026
180cc80
Make some of ColumnReaderImplBase private
AntoinePrv Jul 20, 2026
c8f3c80
Fix horrible dictionnary handling
AntoinePrv Jul 20, 2026
7db94f2
Simplify and rename ColumnChunkReader
AntoinePrv Jul 20, 2026
24ae596
Add RleBitPackedToBitmapDecoder::CountUpTo
AntoinePrv Jul 21, 2026
e9907da
Add BitPackedToBitmapDecoder
AntoinePrv Jul 21, 2026
53466bd
Add RleBitPackedToBitmapDecoder::Advance
AntoinePrv Jul 21, 2026
80cc713
Add LevelToBitmapDecoder
AntoinePrv Jul 21, 2026
878fa3f
Add ValiditySinkBuffer
AntoinePrv Jul 22, 2026
e01f5fa
Auto reserve space in sinks:
AntoinePrv Jul 22, 2026
b352b07
Fix integer types
AntoinePrv Jul 22, 2026
544e4b4
Nit in ColumnChunkReader
AntoinePrv Jul 22, 2026
52dcda9
Eagerly read dict
AntoinePrv Jul 23, 2026
82f1a03
Cleanups
AntoinePrv Jul 24, 2026
144e8fd
Add FlatOptionalTypedRecordReader
AntoinePrv Jul 24, 2026
cc1b422
Add some doc
AntoinePrv Jul 27, 2026
59b612f
Factorize ReadDictionary
AntoinePrv Jul 29, 2026
3377da7
Simplify read_dense_for_nullable
AntoinePrv Jul 29, 2026
3f55b14
Factor Sinks with DataSinkBuffer
AntoinePrv Jul 30, 2026
290adcd
Use LevelBufferSink in TypedRecordReader
AntoinePrv Jul 30, 2026
6c0e310
Relax record reader level tests flat optional
AntoinePrv Jul 30, 2026
83ab573
Fix compiler errrors
AntoinePrv Jul 31, 2026
4966170
Safe decoded number check and func rename
AntoinePrv Jul 31, 2026
1a42d41
Explicit values_written contract
AntoinePrv Jul 31, 2026
7efe675
Fix LevelDecoder::max_level type
AntoinePrv Jul 31, 2026
902379e
Fix integer convertion
AntoinePrv Jul 31, 2026
d83e19e
Factorize level decoders
AntoinePrv Aug 3, 2026
d9cf482
Fixt typos
AntoinePrv Aug 6, 2026
0681d64
Move level decoder utility and rename file
AntoinePrv Sep 29, 2026
bbd546d
Move level decoder adapters to level_decoder.h
AntoinePrv Sep 29, 2026
c757c1e
Fix small typos
AntoinePrv Sep 29, 2026
725da03
Fix issues in level counting
AntoinePrv Sep 29, 2026
1606ccf
Fix extra alloc
AntoinePrv Oct 1, 2026
e33fdc7
Add RecordReader::Make flag
AntoinePrv Oct 6, 2026
608d40a
Throw on FlatOptional def_levels
AntoinePrv Oct 6, 2026
6b8c906
Always call RecordReader with flat optiona optimization
AntoinePrv Oct 6, 2026
dff7065
Addition flat optional test
AntoinePrv Oct 6, 2026
841608f
Detemplate kdic
AntoinePrv Oct 6, 2026
e3a2654
Add flat optiona bse in flba and byte array
AntoinePrv Oct 7, 2026
c44b2eb
Ad more flba and array test
AntoinePrv Oct 7, 2026
352210c
Make RecordReader::Make params struct public
AntoinePrv Oct 8, 2026
a26f8d9
Run RecordReader tests with and without optimizations
AntoinePrv Oct 8, 2026
7823021
Rename flat_optiona_optimization as allow_levels_optimization
AntoinePrv Oct 8, 2026
0df1e9b
Add level optim option to RowGroupReader
AntoinePrv Oct 8, 2026
bca2f09
Reuse validity map
AntoinePrv Oct 13, 2026
7e595f7
Lazily pop dictionnary
AntoinePrv Oct 13, 2026
911a392
Address some review comments
AntoinePrv Oct 21, 2026
3d77c49
Fix typo
AntoinePrv Oct 21, 2026
b4a44de
Add missing gTest PrintTo
AntoinePrv Oct 22, 2026
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
157 changes: 138 additions & 19 deletions cpp/src/arrow/util/rle_bitmap_internal.h
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@

#include <algorithm>
#include <cstring>
#include <limits>

#include "arrow/util/bit_util.h"
#include "arrow/util/bitmap_ops.h"
Expand Down Expand Up @@ -112,6 +113,18 @@ class RleRunToBitmapDecoder {
return n_vals;
}

/// Advance and count the number of occurrences of `value`.
///
/// The count is limited to at most the next `batch_size` items.
/// @return The matching value count and number of elements that were processed.
RleCountUpToResult CountUpTo(bool value, rle_size_t batch_size) {
const auto n_vals = Advance(batch_size);
return {
.matching_count = n_vals * (value == this->value()),
.processed_count = n_vals,
};
}

/// Read the next value into `out` and return false if there are no more.
[[nodiscard]] bool Get(BitmapSpanMut out) { return GetBatch(out, 1) == 1; }

Expand Down Expand Up @@ -229,6 +242,19 @@ class BitPackedRunToBitmapDecoder {
return n_vals;
}

/// Advance and count the number of occurrences of `value`.
///
/// The count is limited to at most the next `batch_size` items.
/// @return The matching value count and number of elements that were processed.
RleCountUpToResult CountUpTo(bool value, rle_size_t batch_size) {
const auto n_vals = GetAdvanceCapacity(batch_size);
const auto set_count = static_cast<rle_size_t>(arrow::internal::CountSetBits(
unread_values_ptr(), unread_values_bit_offset(), n_vals));
const auto matching_count = value ? set_count : n_vals - set_count;
ARROW_UNUSED(Advance(n_vals));
return {.matching_count = matching_count, .processed_count = n_vals};
}

/// Get the next value and return false if there are no more.
[[nodiscard]] bool Get(BitmapSpanMut out) { return GetBatch(out, 1) == 1; }

Expand Down Expand Up @@ -282,11 +308,23 @@ class RleBitPackedToBitmapDecoder {
/// Whether there is still runs to iterate over.
bool exhausted() const { return (run_remaining() == 0) && parser_.exhausted(); }

/// Advance by as many values as provided or until exhaustion of the decoder.
/// Return the number of values skipped.
[[nodiscard]] rle_size_t Advance(rle_size_t batch_size);

/// Get a batch of values return the number of decoded elements.
/// May write fewer elements to the output than requested if there are not enough
/// values left or if an error occurred.
[[nodiscard]] rle_size_t GetBatch(BitmapSpanMut out, rle_size_t batch_size);

/// Advance and count the number of occurrences of `value`.
///
/// The count is limited to at most the next `batch_size` items. Fewer than
/// `batch_size` elements may be processed if there are not enough values left
/// or if an error occurred.
/// @return The matching value count and number of elements that were processed.
RleCountUpToResult CountUpTo(bool value, rle_size_t batch_size);

private:
RleBitPackedParser parser_ = {};
std::variant<RleRunToBitmapDecoder, BitPackedRunToBitmapDecoder> decoder_ = {};
Expand All @@ -296,10 +334,60 @@ class RleBitPackedToBitmapDecoder {
return std::visit([](const auto& dec) { return dec.remaining(); }, decoder_);
}

/// Get a batch of values from the current run and return the number elements read.
[[nodiscard]] rle_size_t RunGetBatch(BitmapSpanMut out, rle_size_t batch_size) {
return std::visit([&](auto& dec) { return dec.GetBatch(out, batch_size); }, decoder_);
/// Process data in the current run and subsequent ones.
///
/// `func` is called as `func(decoder, run_batch_size)` where `decoder` is a
/// statically-typed run decoder (not the variant).
/// Must return the number of values it processed.
///
/// Return the number of values processed.
template <typename Callable>
rle_size_t ProcessValues(Callable&& func, rle_size_t batch_size);
};

/// Minimal decoder for legacy bit packed encoding (BIT_PACKED = 4) into a bitmap.
///
/// This is the bitmap counterpart of ``BitPackedDecoder``: it decodes a single
/// bit-packed run of booleans (each value stored on a single bit) directly into an
/// Arrow validity bitmap, without the RLE framing.
///
/// The number of values that the decoder can represent is up to 2^31 - 1.
class BitPackedToBitmapDecoder : private BitPackedRunToBitmapDecoder {
private:
using Base = BitPackedRunToBitmapDecoder;

public:
using Base::Advance;
using Base::CountUpTo;
using Base::Get;
using Base::GetBatch;
using Base::remaining;

BitPackedToBitmapDecoder() noexcept = default;

/// Create a decoder object.
///
/// @param data and data_size are the raw bytes to decode.
/// @param value_count is the number of values in the run.
BitPackedToBitmapDecoder(const uint8_t* data, rle_size_t data_size,
rle_size_t value_count) noexcept {
Reset(data, data_size, value_count);
}

void Reset(const uint8_t* data, rle_size_t data_size, rle_size_t value_count) noexcept {
ARROW_DCHECK_GE(value_count, 0);
ARROW_DCHECK_LE(value_count, std::numeric_limits<rle_size_t>::max());
const auto run = BitPackedRun{
/* data= */ data,
/* value_count= */ value_count,
/* value_bit_width= */ 1,
/* max_read_bytes= */ data_size,
};
return Base::Reset(run);
}

/// Whether there is still values to iterate over.
bool exhausted() const { return Base::remaining() == 0; }
};

/************************************************
Expand All @@ -320,50 +408,81 @@ struct RleBitPackedToBitmapDecoderGetDecoder<BitPackedRun> {
using type = BitPackedRunToBitmapDecoder;
};

inline auto RleBitPackedToBitmapDecoder::GetBatch(BitmapSpanMut out,
rle_size_t batch_size) -> rle_size_t {
template <typename Callable>
auto RleBitPackedToBitmapDecoder::ProcessValues(Callable&& func,
rle_size_t batch_size) -> rle_size_t {
using ControlFlow = RleBitPackedParser::ControlFlow;

if (ARROW_PREDICT_FALSE(batch_size == 0 || exhausted())) {
return 0;
}

rle_size_t values_read = 0;
rle_size_t values_processed = 0;

// Remaining from a previous call that would have left some unread data from a run.
if (ARROW_PREDICT_FALSE(run_remaining() > 0)) {
const auto read = RunGetBatch(out, batch_size);
values_read += read;
const auto processed =
std::visit([&](auto& decoder) { return func(decoder, batch_size); }, decoder_);
values_processed += processed;

// Either we fulfilled all the batch to be read or we finished remaining run.
if (ARROW_PREDICT_FALSE(values_read == batch_size)) {
return values_read;
if (ARROW_PREDICT_FALSE(values_processed == batch_size)) {
return values_processed;
}
ARROW_DCHECK(run_remaining() == 0);
}

parser_.ParseWithCallable([&](auto run) {
using RunDecoder = RleBitPackedToBitmapDecoderGetDecoder<decltype(run)>::type;

ARROW_DCHECK_LT(values_read, batch_size);
ARROW_DCHECK_LT(values_processed, batch_size);
RunDecoder decoder(run);
// The output span carries its own bit offset, so advancing it past the values
// already written keeps successive runs correctly aligned in the bitmap.
const auto read =
decoder.GetBatch(out.NewStartingAt(values_read), batch_size - values_read);
ARROW_DCHECK_LE(read, batch_size - values_read);
values_read += read;
const auto read = func(decoder, batch_size - values_processed);
ARROW_DCHECK_LE(read, batch_size - values_processed);
values_processed += read;

// Stop reading and store remaining decoder
if (ARROW_PREDICT_FALSE(values_read == batch_size || read == 0)) {
if (ARROW_PREDICT_FALSE(values_processed == batch_size || read == 0)) {
decoder_ = std::move(decoder);
return ControlFlow::Break;
}

return ControlFlow::Continue;
});

return values_read;
return values_processed;
}

inline auto RleBitPackedToBitmapDecoder::Advance(rle_size_t batch_size) -> rle_size_t {
return ProcessValues(
[](auto& decoder, rle_size_t run_batch_size) {
return decoder.Advance(run_batch_size);
},
batch_size);
}

inline auto RleBitPackedToBitmapDecoder::GetBatch(BitmapSpanMut out,
rle_size_t batch_size) -> rle_size_t {
return ProcessValues(
[&out](auto& decoder, rle_size_t run_batch_size) {
const auto read = decoder.GetBatch(out, run_batch_size);
out = out.NewStartingAt(read);
return read;
},
batch_size);
}

inline auto RleBitPackedToBitmapDecoder::CountUpTo(bool value, rle_size_t batch_size)
-> RleCountUpToResult {
rle_size_t matching_count = 0;
const rle_size_t processed_count = ProcessValues(
[value, &matching_count](auto& decoder, rle_size_t run_batch_size) {
const auto result = decoder.CountUpTo(value, run_batch_size);
matching_count += result.matching_count;
return result.processed_count;
},
batch_size);
return {.matching_count = matching_count, .processed_count = processed_count};
}

} // namespace arrow::util
Loading
Loading