Copilot commented on code in PR #50422:
URL: https://github.com/apache/arrow/pull/50422#discussion_r3545170213
##########
cpp/src/arrow/util/rle_encoding_internal.h:
##########
@@ -515,6 +573,71 @@ class RleBitPackedDecoder {
int64_t valid_bits_offset, rle_size_t
null_count);
};
+/// Minimal decoder for legacy bit packed encoding (BIT_PACKED = 4).
+///
+/// The number of values that the decoder can represent is up to 2^31 - 1.
+template <typename T>
+class BitPackedDecoder : private BitPackedRunDecoder<T> {
+ private:
+ using Base = BitPackedRunDecoder<T>;
+
+ public:
+ /// The type in which the data should be decoded.
+ using value_type = T;
+
+ using Base::Advance;
+
+ BitPackedDecoder() noexcept = default;
+
+ /// Create a decoder object.
+ ///
+ /// data and data_size are the raw bytes to decode.
+ /// value_bit_width is the size in bits of each encoded value.
+ BitPackedDecoder(const uint8_t* data, rle_size_t data_size,
+ rle_size_t value_bit_width) noexcept {
+ Reset(data, data_size, value_bit_width);
+ }
+
+ void Reset(const uint8_t* data, rle_size_t data_size,
+ rle_size_t value_bit_width) noexcept {
+ value_bit_width_ = value_bit_width;
+ const auto value_count = (static_cast<int64_t>(data_size) * 8) /
value_bit_width;
+ ARROW_DCHECK_LE(value_count, std::numeric_limits<rle_size_t>::max());
Review Comment:
`BitPackedDecoder::Reset` divides by `value_bit_width` when computing
`value_count`. If `value_bit_width == 0` (e.g. `max_level == 0` => `Log2(1) ==
0`), this is a divide-by-zero and will crash. Please guard `value_bit_width ==
0` and pick a safe `value_count` (the caller already tracks remaining values
separately for level decoding).
##########
cpp/src/parquet/column_reader.cc:
##########
@@ -191,6 +201,22 @@ int LevelDecoder::Decode(int batch_size, int16_t* levels) {
return num_decoded;
}
+int LevelDecoder::Skip(int batch_size) {
+ const int num_values = std::min(num_values_remaining_, batch_size);
+ const int num_advanced = decoder_->Advance(num_values);
+ ARROW_DCHECK_EQ(num_values, num_advanced);
+ num_values_remaining_ -= num_advanced;
+ return num_advanced;
+}
+
+std::pair<int, int> LevelDecoder::CountUpTo(int16_t value, int batch_size) {
+ const int num_values = std::min(num_values_remaining_, batch_size);
+ const auto result = decoder_->CountUpTo(value, num_values);
+ ARROW_DCHECK_EQ(num_values, result.advanced_count);
+ num_values_remaining_ -= result.advanced_count;
Review Comment:
`LevelDecoder::CountUpTo` only DCHECKs that the underlying decoder advanced
`num_values`. In release builds, a short advance would silently corrupt
`num_values_remaining_`. Prefer throwing when `advanced_count != num_values`.
##########
cpp/src/parquet/column_reader.h:
##########
@@ -78,26 +65,44 @@ struct PARQUET_EXPORT DataPageStats {
class PARQUET_EXPORT LevelDecoder {
public:
- LevelDecoder();
+ explicit LevelDecoder(int16_t max_level = 0);
+
~LevelDecoder();
- // Initialize the LevelDecoder state with new data
- // and return the number of bytes consumed
+ /// Initialize the LevelDecoder state with new data from a legacy (V1) page.
+ ///
+ /// @return the number of bytes consumed
int SetData(Encoding::type encoding, int16_t max_level, int
num_buffered_values,
const uint8_t* data, int32_t data_size);
+ /// Initialize the LevelDecoder state with new data from a V2 page.
+ ///
+ /// Encoding of the level in the
Review Comment:
The doc comment for `SetDataV2` is incomplete ("Encoding of the level in
the"). This reads like an unfinished sentence in a public header; please
complete or remove it.
##########
cpp/src/parquet/column_reader.cc:
##########
@@ -191,6 +201,22 @@ int LevelDecoder::Decode(int batch_size, int16_t* levels) {
return num_decoded;
}
+int LevelDecoder::Skip(int batch_size) {
+ const int num_values = std::min(num_values_remaining_, batch_size);
+ const int num_advanced = decoder_->Advance(num_values);
+ ARROW_DCHECK_EQ(num_values, num_advanced);
+ num_values_remaining_ -= num_advanced;
Review Comment:
`LevelDecoder::Skip` only DCHECKs that `Advance()` processed the requested
number of values. In release builds, a short advance would silently
desynchronize `num_values_remaining_` from the underlying decoder and could
cascade into later out-of-bounds reads. Prefer throwing an exception when
`num_advanced != num_values`.
##########
cpp/src/parquet/column_reader.cc:
##########
@@ -1137,40 +1223,40 @@ int64_t TypedColumnReaderImpl<DType>::ReadBatch(int64_t
batch_size, int16_t* def
return total_values;
}
-template <typename DType>
-void TypedColumnReaderImpl<DType>::InitScratchForSkip() {
- if (this->scratch_for_skip_ == nullptr) {
- int value_size = type_traits<DType::type_num>::value_byte_size;
- this->scratch_for_skip_ = AllocateBuffer(
- this->pool_, kSkipScratchBatchSize * std::max<int>(sizeof(int16_t),
value_size));
- }
-}
-
template <typename DType>
int64_t TypedColumnReaderImpl<DType>::Skip(int64_t num_values_to_skip) {
int64_t values_to_skip = num_values_to_skip;
// Optimization: Do not call HasNext() when values_to_skip == 0.
while (values_to_skip > 0 && HasNext()) {
// If the number of values to skip is more than the number of undecoded
values, skip
- // the Page.
+ // the whole Page without decoding levels or values.
const int64_t available_values = this->available_values_current_page();
if (values_to_skip >= available_values) {
values_to_skip -= available_values;
this->ConsumeBufferedValues(available_values);
} else {
- // We need to read this Page
- // Jump to the right offset in the Page
- int64_t values_read = 0;
- InitScratchForSkip();
- ARROW_DCHECK_NE(this->scratch_for_skip_, nullptr);
- do {
- int64_t batch_size = std::min(kSkipScratchBatchSize, values_to_skip);
- values_read = ReadBatch(static_cast<int>(batch_size),
- scratch_for_skip_->mutable_data_as<int16_t>(),
- scratch_for_skip_->mutable_data_as<int16_t>(),
- scratch_for_skip_->mutable_data_as<T>(),
&values_read);
- values_to_skip -= values_read;
- } while (values_read > 0 && values_to_skip > 0);
+ // Skip within the current Page. Since `values_to_skip <
available_values`, the
+ // whole batch fits inside this Page and no page boundary is crossed.
+ const int batch_size = static_cast<int>(values_to_skip);
+
+ // Advance the definition levels, counting how many correspond to present
+ // (non-null) values that must be skipped in the data decoder.
+ int64_t non_null_values_to_skip = batch_size;
+ if (this->max_def_level() > 0) {
+ const auto [count, advanced] =
+ this->definition_level_decoder_.CountUpTo(this->max_def_level(),
batch_size);
+ non_null_values_to_skip = count;
+ ARROW_DCHECK_EQ(advanced, batch_size);
+ }
+ // Advance the repetition levels; their values are not needed.
+ if (this->max_rep_level() > 0) {
+ this->repetition_level_decoder_.Skip(batch_size);
+ }
+ // Skip the corresponding data values.
+ this->current_decoder_.Skip(non_null_values_to_skip, this->pool_);
Review Comment:
In `TypedColumnReaderImpl::Skip`, the code assumes `CountUpTo`,
`LevelDecoder::Skip`, and `SkippableTypedDecoder::Skip` always advance the full
requested counts (only guarded by DCHECKs, and the data decoder return value is
ignored). In release builds, any short advance would still call
`ConsumeBufferedValues(batch_size)`, desynchronizing decoded/available counts
and potentially leading to invalid reads. Please validate the returned advanced
counts and throw on mismatch (similar to other read paths).
##########
cpp/src/parquet/column_reader.cc:
##########
@@ -629,6 +655,70 @@ std::unique_ptr<PageReader>
PageReader::Open(std::shared_ptr<ArrowInputStream> s
namespace {
+/// Wrapper around a `TypedDecoder` pointer to skip values.
+///
+/// Use a scratch buffer to decode values into that buffer.
+/// This was migrated here from a historical implementation.
+/// Ideally all decoders would implement a `Skip` functionality that would at
best
+/// avoid decoding, and at worst, decode without intermediarry allocation.
Review Comment:
Typo in comment: "intermediarry" -> "intermediary".
##########
cpp/src/parquet/column_reader.h:
##########
@@ -78,26 +65,44 @@ struct PARQUET_EXPORT DataPageStats {
class PARQUET_EXPORT LevelDecoder {
public:
- LevelDecoder();
+ explicit LevelDecoder(int16_t max_level = 0);
+
~LevelDecoder();
- // Initialize the LevelDecoder state with new data
- // and return the number of bytes consumed
+ /// Initialize the LevelDecoder state with new data from a legacy (V1) page.
+ ///
+ /// @return the number of bytes consumed
int SetData(Encoding::type encoding, int16_t max_level, int
num_buffered_values,
const uint8_t* data, int32_t data_size);
+ /// Initialize the LevelDecoder state with new data from a V2 page.
+ ///
+ /// Encoding of the level in the
void SetDataV2(int32_t num_bytes, int16_t max_level, int num_buffered_values,
const uint8_t* data);
- // Decodes a batch of levels into an array and returns the number of levels
decoded
+ /// Decode a batch of levels into an array and returns the number of levels
decoded.
int Decode(int batch_size, int16_t* levels);
+ /// Advance the decoder and throw away decoder levels.
+ int Skip(int batch_size);
+
+ /// Advance and count the number of occurrence of a values.
+ ///
+ /// The count is limited to at most the next `batch_size` items.
+ /// @return The matching value count and number of elements that were
processed.
Review Comment:
Grammar in the public `LevelDecoder::CountUpTo` doc comment is incorrect
("occurrence of a values"). Suggest updating to improve API docs.
##########
cpp/src/arrow/util/rle_encoding_internal.h:
##########
@@ -300,6 +313,15 @@ class RleRunDecoder {
return steps;
}
+ /// Advance and count the number of occurrence of a values.
+ ///
+ /// The count is limited to at most the next `batch_size` items.
+ /// @return The matching value count and number of of element that were
processed.
+ RleCountUpToResult CountUpTo(const RleCountUpToParams<value_type>& p) {
Review Comment:
Comment grammar is off in the new `CountUpTo` docs ("occurrence of a
values", "number of of element"). Please fix here and consider updating the
duplicate instances in this file (e.g. the bit-packed decoders) for consistency.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
