Copilot commented on code in PR #50422:
URL: https://github.com/apache/arrow/pull/50422#discussion_r3552797991
##########
cpp/src/parquet/column_reader.cc:
##########
@@ -191,6 +203,25 @@ 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 = impl_->Advance(num_values);
+ ARROW_DCHECK_EQ(num_values, num_advanced);
+ num_values_remaining_ -= num_advanced;
+ return num_advanced;
+}
+
+auto LevelDecoder::CountUpTo(int16_t value, int batch_size) -> CountUpToResult
{
+ const int num_values = std::min(num_values_remaining_, batch_size);
+ const auto result = impl_->CountUpTo(value, num_values);
+ ARROW_DCHECK_EQ(num_values, result.processed_count);
+ num_values_remaining_ -= result.processed_count;
+ return {
+ .matching_count = result.matching_count,
+ .processed_count = result.processed_count,
+ };
+}
Review Comment:
`LevelDecoder::CountUpTo` relies on `ARROW_DCHECK_EQ(num_values,
result.processed_count)`; in release builds a short read (e.g.
corrupt/truncated page) would silently reduce `num_values_remaining_` and
return partial counts, desynchronizing downstream decoding. This should raise
an error when fewer than `num_values` elements are processed.
##########
cpp/src/parquet/column_reader.cc:
##########
@@ -191,6 +203,25 @@ 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 = impl_->Advance(num_values);
+ ARROW_DCHECK_EQ(num_values, num_advanced);
+ num_values_remaining_ -= num_advanced;
+ return num_advanced;
+}
Review Comment:
`LevelDecoder::Skip` only enforces that the underlying decoder advanced the
requested number of values with `ARROW_DCHECK_EQ`, which is compiled out in
release builds. If the page is truncated/corrupt and `Advance()` returns fewer
values than `num_values_remaining_` indicates, the reader will silently
desynchronize instead of raising an error.
##########
cpp/src/parquet/column_reader.cc:
##########
@@ -1137,40 +1229,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 =
+ this->definition_level_decoder_.CountUpTo(this->max_def_level(),
batch_size);
+ non_null_values_to_skip = count.matching_count;
+ ARROW_DCHECK_EQ(count.processed_count, 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->ConsumeBufferedValues(batch_size);
Review Comment:
`TypedColumnReaderImpl::Skip` ignores the return value of
`current_decoder_.Skip(non_null_values_to_skip)`. If the underlying decoder
fails to skip the requested number of non-null values (e.g. due to
truncated/corrupt data), the reader will still
`ConsumeBufferedValues(batch_size)` and silently desynchronize.
--
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]
