This is an automated email from the ASF dual-hosted git repository.
apitrou pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/arrow.git
The following commit(s) were added to refs/heads/master by this push:
new 4bc9355020 PARQUET-2204: [parquet-cpp] TypedColumnReaderImpl::Skip
should reuse scratch space (#14509)
4bc9355020 is described below
commit 4bc935502065b490b6be1d3201fb3e134a92dcb3
Author: Fatemah Panahi <[email protected]>
AuthorDate: Thu Dec 8 07:42:00 2022 -0800
PARQUET-2204: [parquet-cpp] TypedColumnReaderImpl::Skip should reuse
scratch space (#14509)
TypedColumnReaderImpl::Skip allocates scratch space on every call. The
scratch space is used to read rep/def levels and values and throw them away.
The memory allocation slows down the skip based on microbenchmarks. The scratch
space can be allocated once and re-used.
Lead-authored-by: Fatemah Panahi <[email protected]>
Co-authored-by: Fatemah Panahi <[email protected]>
Signed-off-by: Antoine Pitrou <[email protected]>
---
cpp/src/parquet/column_reader.cc | 58 ++++++++++++++++++++++++----------------
1 file changed, 35 insertions(+), 23 deletions(-)
diff --git a/cpp/src/parquet/column_reader.cc b/cpp/src/parquet/column_reader.cc
index cab9bb0fb3..dc98041d3a 100644
--- a/cpp/src/parquet/column_reader.cc
+++ b/cpp/src/parquet/column_reader.cc
@@ -68,6 +68,10 @@ namespace {
// better vectorized performance when doing many smaller record reads
constexpr int64_t kMinLevelBatchSize = 1024;
+// Batch size for reading and throwing away values during skip.
+// Both RecordReader and the ColumnReader use this for skipping.
+constexpr int64_t kSkipScratchBatchSize = 1024;
+
inline bool HasSpacedValues(const ColumnDescriptor* descr) {
if (descr->max_repetition_level() > 0) {
// repeated+flat case
@@ -941,6 +945,14 @@ class TypedColumnReaderImpl : public
TypedColumnReader<DType>,
this->exposed_encoding_ = encoding;
}
+ // Allocate enough scratch space to accommodate skipping 16-bit levels or any
+ // value type.
+ void InitScratchForSkip();
+
+ // Scratch space for reading and throwing away rep/def levels and values when
+ // skipping.
+ std::shared_ptr<ResizableBuffer> scratch_for_skip_;
+
private:
// Read dictionary indices. Similar to ReadValues but decode data to
dictionary indices.
// This function is called only by ReadBatchWithDictionary().
@@ -1151,6 +1163,15 @@ int64_t TypedColumnReaderImpl<DType>::ReadBatchSpaced(
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;
@@ -1165,23 +1186,16 @@ int64_t TypedColumnReaderImpl<DType>::Skip(int64_t
num_values_to_skip) {
} else {
// We need to read this Page
// Jump to the right offset in the Page
- int64_t batch_size =
- kMinLevelBatchSize; // ReadBatch with a smaller memory footprint
int64_t values_read = 0;
-
- // This will be enough scratch space to accommodate 16-bit levels or any
- // value type
- int value_size = type_traits<DType::type_num>::value_byte_size;
- std::shared_ptr<ResizableBuffer> scratch = AllocateBuffer(
- this->pool_, batch_size * std::max<int>(sizeof(int16_t),
value_size));
-
+ InitScratchForSkip();
+ ARROW_DCHECK_NE(this->scratch_for_skip_, nullptr);
do {
- batch_size = std::min(batch_size, values_to_skip);
- values_read =
- ReadBatch(static_cast<int>(batch_size),
- reinterpret_cast<int16_t*>(scratch->mutable_data()),
- reinterpret_cast<int16_t*>(scratch->mutable_data()),
- reinterpret_cast<T*>(scratch->mutable_data()),
&values_read);
+ int64_t batch_size = std::min(kSkipScratchBatchSize, values_to_skip);
+ values_read = ReadBatch(
+ static_cast<int>(batch_size),
+
reinterpret_cast<int16_t*>(this->scratch_for_skip_->mutable_data()),
+
reinterpret_cast<int16_t*>(this->scratch_for_skip_->mutable_data()),
+ reinterpret_cast<T*>(this->scratch_for_skip_->mutable_data()),
&values_read);
values_to_skip -= values_read;
} while (values_read > 0 && values_to_skip > 0);
}
@@ -1517,18 +1531,16 @@ class TypedRecordReader : public
TypedColumnReaderImpl<DType>,
// Throws an error if it could not read 'num_values'.
void ReadAndThrowAwayValues(int64_t num_values) {
int64_t values_left = num_values;
- int64_t batch_size = kMinLevelBatchSize; // ReadBatch with a smaller
memory footprint
int64_t values_read = 0;
- // This will be enough scratch space to accommodate 16-bit levels or any
+ // Allocate enough scratch space to accommodate 16-bit levels or any
// value type
- int value_size = type_traits<DType::type_num>::value_byte_size;
- std::shared_ptr<ResizableBuffer> scratch = AllocateBuffer(
- this->pool_, batch_size * std::max<int>(sizeof(int16_t), value_size));
+ this->InitScratchForSkip();
+ ARROW_DCHECK_NE(this->scratch_for_skip_, nullptr);
do {
- batch_size = std::min<int64_t>(batch_size, values_left);
- values_read =
- this->ReadValues(batch_size,
reinterpret_cast<T*>(scratch->mutable_data()));
+ int64_t batch_size = std::min<int64_t>(kSkipScratchBatchSize,
values_left);
+ values_read = this->ReadValues(
+ batch_size,
reinterpret_cast<T*>(this->scratch_for_skip_->mutable_data()));
values_left -= values_read;
} while (values_read > 0 && values_left > 0);
if (values_left > 0) {
