Copilot commented on code in PR #50422:
URL: https://github.com/apache/arrow/pull/50422#discussion_r3550231459
##########
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());
+ const auto run = BitPackedRun{
+ /* data= */ data,
+ /* value_count= */ static_cast<rle_size_t>(value_count),
+ /* value_bit_width= */ value_bit_width,
+ };
+ return Base::Reset(run, value_bit_width);
+ }
Review Comment:
`BitPackedDecoder::Reset` computes `value_count` by dividing by
`value_bit_width`, which can be 0 (e.g. max_level == 0 => Log2(max_level + 1)
== 0). This is a divide-by-zero and will crash when initializing a BIT_PACKED
levels decoder.
##########
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;
+}
Review Comment:
`LevelDecoder::Skip` only DCHECKs that the underlying decoder advanced the
requested number of values. In non-DCHECK builds, truncated/corrupt level data
could result in skipping fewer levels than requested while the caller proceeds
as if it skipped everything (e.g. `TypedColumnReaderImpl::Skip`). Consider
throwing a `ParquetException` on mismatch to match the existing "rep/def levels
not match" error handling used for Decode paths.
##########
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;
+ return {result.count, result.advanced_count};
+}
Review Comment:
`LevelDecoder::CountUpTo` only DCHECKs that the underlying decoder advanced
`num_values`. In release builds, corrupted/truncated data could return a
smaller `advanced_count` while downstream code assumes a full advance (and
consumes `batch_size` header values). Throwing on mismatch would prevent silent
desynchronization.
##########
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_);
+
+ this->ConsumeBufferedValues(batch_size);
Review Comment:
In the within-page skip path, the return value of
`current_decoder_.Skip(...)` is ignored. If the data decoder advances fewer
than `non_null_values_to_skip` (e.g. truncated/corrupt value buffer),
`ConsumeBufferedValues(batch_size)` will still advance the page position,
potentially desynchronizing levels vs values. Since this branch guarantees the
skip stays within the current page, it should validate the data decoder
advanced the expected number of physical values (and throw on mismatch).
--
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]
