apache / apache/arrow

[C++] Lengthy destruction of ScannerRecordBatchReader

オープン
#40,653 コメント 3 件 リアクション 0 件 担当者 1 名 @pitrou が担当を希望しています GitHub で見る
Component: C++ Type: bug
主要言語
C++
スター
17.1k
フォーク
4.3k
平均マージ
3日 13時間
マージ済み PR(30日)
88

説明

### Describe the bug, including details regarding any error messages, version, and platform.

The destruction of the ScannerRecordBatchReader object returned by arrow::dataset::Scanner::ToRecordBatchReader() on a Parquet dataset of 1 GB with ~ 10 million rows and 77 row groups (https://overturemaps-us-west-2.s3.amazonaws.com/release/2024-03-12-alpha.0/theme%3Dbuildings/type%3Dbuilding/part-00000-4dfc75cd-2680-4d52-b5e0-f4cc9f36b267-c000.zstd.parquet) is extremely long, when reading for example just only a few rows, due to SerialIterator::~SerialIterator() iterating until the end of the dataset. It would be desirable that the destruction of the batch reader doesn't trigger such lengthy operations.

thread_pool.h has the following comment, but trying to implement that is beyond my understanding of the libarrow/libparquet deep internals:
```
/// Note: The iterator's destructor will run until the given generator is fully
/// exhausted. If you wish to abandon iteration before completion then the correct
/// approach is to use a stop token to cause the generator to exhaust early.
```

Invoking explicitly the Close() method on the record batch reader doesn't improve performance either.

This is the result of the analysis of https://github.com/OSGeo/gdal/issues/9497

Version: libarrow/libparquet from apache-arrow-15.0.0

Related stack trace:
```
#0 arrow::io::RandomAccessFile::ReadAsync (this=0x555555fb42e0, ctx=..., position=31630301, nbytes=29124528) at /home/even/arrow/cpp/src/arrow/io/interfaces.cc:169
#1 0x00007fffdf8eb916 in arrow::io::internal::ReadRangeCache::LazyImpl::MaybeRead (this=0x7fffd41ffe00, entry=0x7fffd420b3e0) at /home/even/arrow/cpp/src/arrow/io/caching.cc:270
#2 0x00007fffdf8eb74b in arrow::io::internal::ReadRangeCache::Impl::WaitFor (this=0x7fffd41ffe00, ranges=std::vector of length 33, capacity 33 = {...}) at /home/even/arrow/cpp/src/arrow/io/caching.cc:249
#3 0x00007fffdf8ebd8f in arrow::io::internal::ReadRangeCache::LazyImpl::WaitFor (this=0x7fffd41ffe00, ranges=std::vector of length 0, capacity 0) at /home/even/arrow/cpp/src/arrow/io/caching.cc:304
#4 0x00007fffdf8ea26c in arrow::io::internal::ReadRangeCache::WaitFor (this=0x7fffd4200900, ranges=std::vector of length 0, capacity 0) at /home/even/arrow/cpp/src/arrow/io/caching.cc:331
#5 0x00007fffec678c0f in parquet::SerializedFile::WhenBuffered (this=0x555556014860, row_groups=std::vector of length 1, capacity 1 = {...}, column_indices=std::vector of length 33, capacity 33 = {...}) at /home/even/arrow/cpp/src/parquet/file_reader.cc:418
#6 0x00007fffec6750f3 in parquet::ParquetFileReader::WhenBuffered (this=0x7fffd41f52d0, row_groups=std::vector of length 1, capacity 1 = {...}, column_indices=std::vector of length 33, capacity 33 = {...}) at /home/even/arrow/cpp/src/parquet/file_reader.cc:905
#7 0x00007fffec41d812 in parquet::arrow::RowGroupGenerator::FetchNext (this=0x7fffd4001630) at /home/even/arrow/cpp/src/parquet/arrow/reader.cc:1123
#8 0x00007fffec41d342 in parquet::arrow::RowGroupGenerator::FillReadahead (this=0x7fffd4001630) at /home/even/arrow/cpp/src/parquet/arrow/reader.cc:1102
#9 0x00007fffec41d1ca in parquet::arrow::RowGroupGenerator::operator() (this=0x7fffd4001630) at /home/even/arrow/cpp/src/parquet/arrow/reader.cc:1090
#10 0x00007fffec42da9c in std::_Function_handler > ()> > (), parquet::arrow::RowGroupGenerator>::_M_invoke(std::_Any_data const&) (__functor=...) at /usr/include/c++/9/bits/std_function.h:286
#11 0x00007fffecfce934 in std::function > ()> > ()>::operator()() const (this=0x7fffd42026a0) at /usr/include/c++/9/bits/std_function.h:688
#12 0x00007fffecfcca75 in arrow::MergedGenerator >::State::PullSource (this=0x7fffd42026a0) at /home/even/arrow/cpp/src/arrow/util/async_generator.h:1158
#13 0x00007fffecfdb2e1 in arrow::MergedGenerator >::InnerCallback::operator() (this=0x555557324588, maybe_next_ref=...) at /home/even/arrow/cpp/src/arrow/util/async_generator.h:1335
#14 0x00007fffecfd9dfb in arrow::Future >::WrapResultOnComplete::Callback >::InnerCallback>::operator()(arrow::FutureImpl const&) && (this=0x555557324588, impl=...) at /home/even/arrow/cpp/src/arrow/util/future.h:442
#15 0x00007fffecfd94f1 in arrow::internal::FnOnce::FnImpl >::WrapResultOnComplete::Callback >::InnerCallback> >::invoke(arrow::FutureImpl const&) (this=0x555557324580, a#0=...) at /home/even/arrow/cpp/src/arrow/util/functional.h:152
#16 0x00007fffdf9ba13c in arrow::internal::FnOnce::operator()(arrow::FutureImpl const&) && (this=0x7fffffffb7b0, a#0=...) at /home/even/arrow/cpp/src/arrow/util/functional.h:140
#17 0x00007fffdf9b97ea in arrow::ConcreteFutureImpl::RunOrScheduleCallback (self=std::shared_ptr (use count 2, weak count 1) = {...}, callback_record=..., in_add_callback=true) at /home/even/arrow/cpp/src/arrow/util/future.cc:110
#18 0x00007fffdf9b9224 in arrow::ConcreteFutureImpl::AddCallback(arrow::internal::FnOnce, arrow::CallbackOptions) (this=0x5555573d4400, callback=..., opts=...) at /home/even/arrow/cpp/src/arrow/util/future.cc:64
#19 0x00007fffdf9b7112 in arrow::FutureImpl::AddCallback(arrow::internal::FnOnce, arrow::CallbackOptions) (this=0x5555573d4400, callback=..., opts=...) at /home/even/arrow/cpp/src/arrow/util/future.cc:229
#20 0x00007fffecfcc857 in arrow::Future >::AddCallback >::InnerCallback, arrow::Future >::WrapResultOnComplete::Callback >::InnerCallback> > (this=0x7fffffffb900, on_complete=..., opts=...) at /home/even/arrow/cpp/src/arrow/util/future.h:493
#21 0x00007fffecfca507 in arrow::MergedGenerator >::operator() (this=0x7fffd4200610) at /home/even/arrow/cpp/src/arrow/util/async_generator.h:1086
#22 0x00007fffec436eff in std::_Function_handler > (), arrow::MergedGenerator > >::_M_invoke(std::_Any_data const&) (__functor=...) at /usr/include/c++/9/bits/std_function.h:286
#23 0x00007fffecfcaeb2 in std::function > ()>::operator()() const (this=0x7fffd4202a60) at /usr/include/c++/9/bits/std_function.h:688
#24 0x00007fffed1034e6 in arrow::dataset::SlicingGenerator::operator() (this=0x7fffd42005f0) at /home/even/arrow/cpp/src/arrow/dataset/file_parquet.cc:564
#25 0x00007fffed10b8fd in std::_Function_handler > (), arrow::dataset::SlicingGenerator>::_M_invoke(std::_Any_data const&) (__functor=...) at /usr/include/c++/9/bits/std_function.h:286
#26 0x00007fffecfcaeb2 in std::function > ()>::operator()() const (this=0x555555f9b5f0) at /usr/include/c++/9/bits/std_function.h:688
#27 0x00007fffecfc8c07 in arrow::FutureFirstGenerator >::operator() (this=0x555555bc34c0) at /home/even/arrow/cpp/src/arrow/util/async_generator.h:665
#28 0x00007fffecfc6c55 in std::_Function_handler > (), arrow::FutureFirstGenerator > >::_M_invoke(std::_Any_data const&) (__functor=...) at /usr/include/c++/9/bits/std_function.h:286
#29 0x00007fffecfcaeb2 in std::function > ()>::operator()() const (this=0x555555a154d0) at /usr/include/c++/9/bits/std_function.h:688
#30 0x00007fffed051d3a in arrow::DefaultIfEmptyGenerator >::operator() (this=0x7fffb4061370) at /home/even/arrow/cpp/src/arrow/util/async_generator.h:2036
#31 0x00007fffed047b49 in std::_Function_handler > (), arrow::DefaultIfEmptyGenerator > >::_M_invoke(std::_Any_data const&) (__functor=...) at /usr/include/c++/9/bits/std_function.h:286
#32 0x00007fffecfcaeb2 in std::function > ()>::operator()() const (this=0x7fffb40513d0) at /usr/include/c++/9/bits/std_function.h:688
#33 0x00007fffed052222 in arrow::EnumeratingGenerator >::operator() (this=0x7fffb40616f0) at /home/even/arrow/cpp/src/arrow/util/async_generator.h:1524
#34 0x00007fffed047dba in std::_Function_handler > > (), arrow::EnumeratingGenerator > >::_M_invoke(std::_Any_data const&) (__functor=...) at /usr/include/c++/9/bits/std_function.h:286
#35 0x00007fffed05cf20 in std::function > > ()>::operator()() const (this=0x555555a15580) at /usr/include/c++/9/bits/std_function.h:688
#36 0x00007fffed052777 in arrow::FutureFirstGenerator > >::operator() (this=0x555555ea1250) at /home/even/arrow/cpp/src/arrow/util/async_generator.h:665
#37 0x00007fffed048210 in std::_Function_handler > > (), arrow::FutureFirstGenerator > > >::_M_invoke(std::_Any_data const&) (__functor=...) at /usr/include/c++/9/bits/std_function.h:286
#38 0x00007fffed05cf20 in std::function > > ()>::operator()() const (this=0x555555bc2ee0) at /usr/include/c++/9/bits/std_function.h:688
#39 0x00007fffed052cb2 in arrow::MappingGenerator >, arrow::dataset::EnumeratedRecordBatch>::operator() (this=0x55555775b360) at /home/even/arrow/cpp/src/arrow/util/async_generator.h:163
#40 0x00007fffed0484d5 in std::_Function_handler (), arrow::MappingGenerator >, arrow::dataset::EnumeratedRecordBatch> >::_M_invoke(std::_Any_data const&) (__functor=...) at /usr/include/c++/9/bits/std_function.h:286
#41 0x00007fffed033f92 in std::function ()>::operator()() const (this=0x5555578691a0) at /usr/include/c++/9/bits/std_function.h:688
#42 0x00007fffed059b82 in arrow::MergedGenerator::operator() (this=0x555555fb3a60) at /home/even/arrow/cpp/src/arrow/util/async_generator.h:1086
#43 0x00007fffed04f29a in std::_Function_handler (), arrow::MergedGenerator >::_M_invoke(std::_Any_data const&) (__functor=...) at /usr/include/c++/9/bits/std_function.h:286
#44 0x00007fffed033f92 in std::function ()>::operator()() const (this=0x55555598f5a0) at /usr/include/c++/9/bits/std_function.h:688
#45 0x00007fffed05ac48 in arrow::MappingGenerator >::operator() (this=0x555555c166e0) at /home/even/arrow/cpp/src/arrow/util/async_generator.h:163
#46 0x00007fffed04fc1d in std::_Function_handler > (), arrow::MappingGenerator > >::_M_invoke(std::_Any_data const&) (__functor=...) at /usr/include/c++/9/bits/std_function.h:286
#47 0x00007fffed05f73e in std::function > ()>::operator()() const (this=0x5555561360c0) at /usr/include/c++/9/bits/std_function.h:688
#48 0x00007fffe6545d41 in arrow::acero::(anonymous namespace)::SourceNode::::operator()(void) const (__closure=0x555555ea0e28) at /home/even/arrow/cpp/src/arrow/acero/source_node.cc:201
#49 0x00007fffe6549b77 in arrow::Callback::operator() (this=0x555555ea0e28, maybe_control=...) at /home/even/arrow/cpp/src/arrow/util/future.h:837
#50 0x00007fffe6551051 in arrow::Future >::WrapResultOnComplete::Callback; Control = std::optional; BreakValueType = int]::Callback>::operator()(const arrow::FutureImpl &) (this=0x555555ea0e28, impl=...) at /home/even/arrow/cpp/src/arrow/util/future.h:442
#51 0x00007fffe6550f39 in arrow::internal::FnOnce::FnImpl >::WrapResultOnComplete::Callback; Control = std::optional; BreakValueType = int]::Callback> >::invoke(const arrow::FutureImpl &) (this=0x555555ea0e20, a#0=...) at /home/even/arrow/cpp/src/arrow/util/functional.h:152
#52 0x00007fffdf9ba13c in arrow::internal::FnOnce::operator()(arrow::FutureImpl const&) && (this=0x5555578d3e70, a#0=...) at /home/even/arrow/cpp/src/arrow/util/functional.h:140
#53 0x00007fffdf9b97ea in arrow::ConcreteFutureImpl::RunOrScheduleCallback (self=std::shared_ptr (use count 2, weak count 1) = {...}, callback_record=..., in_add_callback=false) at /home/even/arrow/cpp/src/arrow/util/future.cc:110
#54 0x00007fffdf9b9b4e in arrow::ConcreteFutureImpl::DoMarkFinishedOrFailed (this=0x5555578d4490, state=arrow::FutureState::SUCCESS) at /home/even/arrow/cpp/src/arrow/util/future.cc:148
#55 0x00007fffdf9b8fc5 in arrow::ConcreteFutureImpl::DoMarkFinished (this=0x5555578d4490) at /home/even/arrow/cpp/src/arrow/util/future.cc:39
#56 0x00007fffdf9b7072 in arrow::FutureImpl::MarkFinished (this=0x5555578d4490) at /home/even/arrow/cpp/src/arrow/util/future.cc:224
#57 0x00007fffe65593ae in arrow::Future >::DoMarkFinished (this=0x5555565a5b48, res=...) at /home/even/arrow/cpp/src/arrow/util/future.h:658
#58 0x00007fffe65591c9 in arrow::Future >::MarkFinished (this=0x5555565a5b48, res=...) at /home/even/arrow/cpp/src/arrow/util/future.h:403
#59 0x00007fffe65595c6 in arrow::detail::MarkNextFinished >, arrow::Future >, false, false>::operator()(arrow::Result > const&) && (this=0x5555565a5b48, res=...) at /home/even/arrow/cpp/src/arrow/util/future.h:111
#60 0x00007fffe6559573 in arrow::Future >::WrapResultOnComplete::Callback >, arrow::Future >, false, false> >::operator()(arrow::FutureImpl const&) && (this=0x5555565a5b48, impl=...) at /home/even/arrow/cpp/src/arrow/util/future.h:442
#61 0x00007fffe655952f in arrow::internal::FnOnce::FnImpl >::WrapResultOnComplete::Callback >, arrow::Future >, false, false> > >::invoke(arrow::FutureImpl const&) (this=0x5555565a5b40, a#0=...) at /home/even/arrow/cpp/src/arrow/util/functional.h:152
#62 0x00007fffdf9ba13c in arrow::internal::FnOnce::operator()(arrow::FutureImpl const&) && (this=0x7fffffffc7b0, a#0=...) at /home/even/arrow/cpp/src/arrow/util/functional.h:140
#63 0x00007fffdf9b97ea in arrow::ConcreteFutureImpl::RunOrScheduleCallback (self=std::shared_ptr (use count 2, weak count 1) = {...}, callback_record=..., in_add_callback=true) at /home/even/arrow/cpp/src/arrow/util/future.cc:110
#64 0x00007fffdf9b9224 in arrow::ConcreteFutureImpl::AddCallback(arrow::internal::FnOnce, arrow::CallbackOptions) (this=0x555557405e20, callback=..., opts=...) at /home/even/arrow/cpp/src/arrow/util/future.cc:64
#65 0x00007fffdf9b7112 in arrow::FutureImpl::AddCallback(arrow::internal::FnOnce, arrow::CallbackOptions) (this=0x555557405e20, callback=..., opts=...) at /home/even/arrow/cpp/src/arrow/util/future.cc:229
#66 0x00007fffe655910b in arrow::Future >::AddCallback >, arrow::Future >, false, false>, arrow::Future >::WrapResultOnComplete::Callback >, arrow::Future >, false, false> > > (this=0x7fffffffc8d0, on_complete=..., opts=...) at /home/even/arrow/cpp/src/arrow/util/future.h:493
#67 0x00007fffe6551877 in arrow::detail::ContinueFuture::operator()::&)>, const std::optional&>(arrow::Future >, arrow::acero::(anonymous namespace)::SourceNode::::&)> &&) const (this=0x7fffffffc9bf, next=..., f=...) at /home/even/arrow/cpp/src/arrow/util/future.h:181
#68 0x00007fffe655149a in arrow::detail::ContinueFuture::IgnoringArgsIf::&)>, arrow::Future >, const std::optional&>(std::false_type, arrow::Future > &&, arrow::acero::(anonymous namespace)::SourceNode::::&)> &&) const (this=0x7fffffffc9bf, next=..., f=...) at /home/even/arrow/cpp/src/arrow/util/future.h:193
#69 0x00007fffe655120c in arrow::Future >::ThenOnComplete::&)>, arrow::acero::(anonymous namespace)::SourceNode::StartProducing():::: >::operator()(const arrow::Result > &) (this=0x555555ea0518, result=...) at /home/even/arrow/cpp/src/arrow/util/future.h:545
#70 0x00007fffe6551091 in arrow::Future >::WrapResultOnComplete::Callback >::ThenOnComplete::&)>, arrow::acero::(anonymous namespace)::SourceNode::StartProducing():::: > >::operator()(const arrow::FutureImpl &) (this=0x555555ea0518, impl=...) at /home/even/arrow/cpp/src/arrow/util/future.h:442
#71 0x00007fffe6550f81 in arrow::internal::FnOnce::FnImpl >::WrapResultOnComplete::Callback >::ThenOnComplete::&)>, arrow::acero::(anonymous namespace)::SourceNode::StartProducing():::: > > >::invoke(const arrow::FutureImpl &) (this=0x555555ea0510, a#0=...) at /home/even/arrow/cpp/src/arrow/util/functional.h:152
#72 0x00007fffdf9ba13c in arrow::internal::FnOnce::operator()(arrow::FutureImpl const&) && (this=0x7fffb41af678, a#0=...) at /home/even/arrow/cpp/src/arrow/util/functional.h:140
#73 0x00007fffdf9b956e in arrow::ConcreteFutureImpl::RunOrScheduleCallback(std::shared_ptr const&, arrow::FutureImpl::CallbackRecord&&, bool)::{lambda()#1}::operator()() (__closure=0x7fffb41af668) at /home/even/arrow/cpp/src/arrow/util/future.cc:106
#74 0x00007fffdf9bf88a in arrow::internal::FnOnce::FnImpl const&, arrow::FutureImpl::CallbackRecord&&, bool)::{lambda()#1}>::invoke() (this=0x7fffb41af660) at /home/even/arrow/cpp/src/arrow/util/functional.h:152
#75 0x00007fffdfa19f7d in arrow::internal::FnOnce::operator()() && (this=0x7fffffffcb70) at /home/even/arrow/cpp/src/arrow/util/functional.h:140
#76 0x00007fffdfa146ad in arrow::internal::SerialExecutor::RunLoop (this=0x55555611d5b0) at /home/even/arrow/cpp/src/arrow/util/thread_pool.cc:252
#77 0x00007fffed03d806 in arrow::internal::SerialExecutor::IterateGenerator(arrow::internal::FnOnce ()> > (arrow::internal::Executor*)>)::SerialIterator::Next() (this=0x555555a15110) at /home/even/arrow/cpp/src/arrow/util/thread_pool.h:363
#78 0x00007fffed03d6e2 in arrow::internal::SerialExecutor::IterateGenerator(arrow::internal::FnOnce ()> > (arrow::internal::Executor*)>)::SerialIterator::~SerialIterator() (this=0x555555a15110, __in_chrg=) at /home/even/arrow/cpp/src/arrow/util/thread_pool.h:338
#79 0x00007fffed055120 in arrow::Iterator::Delete(arrow::internal::FnOnce ()> > (arrow::internal::Executor*)>)::SerialIterator>(void*) (ptr=0x555555a15110) at /home/even/arrow/cpp/src/arrow/util/iterator.h:193
#80 0x00007fffecf3284c in std::unique_ptr::~unique_ptr (this=0x555555a152b8, __in_chrg=) at /usr/include/c++/9/bits/unique_ptr.h:292
#81 0x00007fffed02e578 in arrow::Iterator::~Iterator (this=0x555555a152b8, __in_chrg=) at /home/even/arrow/cpp/src/arrow/util/iterator.h:87
#82 0x00007fffed0288f4 in arrow::dataset::(anonymous namespace)::ScannerRecordBatchReader::~ScannerRecordBatchReader (this=0x555555a152a0, __in_chrg=) at /home/even/arrow/cpp/src/arrow/dataset/scanner.cc:96
#83 0x00007fffed02aded in __gnu_cxx::new_allocator::destroy (this=0x555555a152a0, __p=0x555555a152a0) at /usr/include/c++/9/ext/new_allocator.h:152
#84 0x00007fffed02a189 in std::allocator_traits >::destroy (__a=..., __p=0x555555a152a0) at /usr/include/c++/9/bits/alloc_traits.h:496
#85 0x00007fffed0296c9 in std::_Sp_counted_ptr_inplace, (__gnu_cxx::_Lock_policy)2>::_M_dispose (this=0x555555a15290) at /usr/include/c++/9/bits/shared_ptr_base.h:557
#86 0x00007fffeda2d84c in std::_Sp_counted_base<(__gnu_cxx::_Lock_policy)2>::_M_release (this=0x555555a15290) at /usr/include/c++/9/bits/shared_ptr_base.h:148
#87 std::_Sp_counted_base<(__gnu_cxx::_Lock_policy)2>::_M_release (this=0x555555a15290) at /usr/include/c++/9/bits/shared_ptr_base.h:148
#88 std::__shared_count<(__gnu_cxx::_Lock_policy)2>::~__shared_count (this=, __in_chrg=) at /usr/include/c++/9/bits/shared_ptr_base.h:730
#89 std::__shared_ptr::~__shared_ptr (this=, __in_chrg=) at /usr/include/c++/9/bits/shared_ptr_base.h:1169
#90 std::__shared_ptr::reset (this=0x555555993288) at /usr/include/c++/9/bits/shared_ptr_base.h:1287
#91 OGRParquetDatasetLayer::ResetReading (this=0x555555992ec0) at /home/even/gdal/gdal/ogr/ogrsf_frmts/parquet/ogrparquetdatasetlayer.cpp:104
```

### Component(s)

C++

コントリビューションガイド

コントリビューションガイドを開く

評価

この issue はまだ評価されていません。

新しい issue をメールで受け取る

初心者向けの GitHub issue を短くまとめたダイジェスト。