You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by jo...@apache.org on 2021/04/26 15:46:46 UTC
[impala] 02/04: IMPALA-10584: Defer advancing read page if stream
only has 2 pages.
This is an automated email from the ASF dual-hosted git repository.
joemcdonnell pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 91d2ab2116293b8f45afd02d029d47233877c538
Author: Riza Suminto <ri...@cloudera.com>
AuthorDate: Sun Apr 11 23:10:57 2021 -0700
IMPALA-10584: Defer advancing read page if stream only has 2 pages.
TestScratchLimit::test_with_unlimited_scratch_limit has been
intermittently crashing in ubuntu-16.04-dockerised-tests environment
after result spooling is enabled by default in IMPALA-9856. DCHECK
violation occurs in ReservationTracker::CheckConsistency() due to
BufferedTupleStream wrongly tries to reclaim memory reservation while
unpinning the stream.
For this bug to surface, all of the following needs to happen:
- Stream is in pinned mode.
- There are only 2 pages in the stream: 1 read and 1 write.
- Stream can not increase reservation anymore either due to memory
pressure or low buffer/memory limit.
- The stream read page has been fully read and is attached to output
RowBatch. But the output RowBatch has not cleaned up yet.
- BufferedTupleStream::UnpinStream is invoked.
The memory accounting bug happens because UnpinStream proceeds to
NextReadPage where the read page buffer was mistakenly assumed as
released. default_page_len_ bytes were added into
write_page_reservation_ and subsequently violates the total memory
reservation.
This patch fixes the bug by deferring advancement of the read iterator
in UnpinStream if the read page is attached to output RowBatch and there
are only 2 pages in the stream. This is OK because after UnpinStream
finished, the stream is now in unpinned mode and has_read_write_page is
false. The next AddRow operation is then allowed to unpin the previous
write page first before reusing the reservation to allocate a new write
page. The next GetNext call will be responsible to advance the read
page.
Testing:
- Add be test DeferAdvancingReadPage.
- Loop the TestScratchLimit::test_with_unlimited_scratch_limit in my
local dev machine and verify that each test passed without triggering
the DCHECK violation.
- Reenable result spooling in TestScratchLimit that was disabled in
IMPALA-10559.
- Pass core tests.
Change-Id: I16137b6e423f190f60c3115a06ccd0f77e9f585a
Reviewed-on: http://gerrit.cloudera.org:8080/17195
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
be/src/runtime/buffered-tuple-stream-test.cc | 70 ++++++++++++++++++++++++++++
be/src/runtime/buffered-tuple-stream.cc | 31 +++++++++++-
be/src/runtime/buffered-tuple-stream.h | 1 +
tests/query_test/test_scratch_limit.py | 6 ---
4 files changed, 100 insertions(+), 8 deletions(-)
diff --git a/be/src/runtime/buffered-tuple-stream-test.cc b/be/src/runtime/buffered-tuple-stream-test.cc
index 2a0d607..dfdfeba 100644
--- a/be/src/runtime/buffered-tuple-stream-test.cc
+++ b/be/src/runtime/buffered-tuple-stream-test.cc
@@ -1297,6 +1297,76 @@ TEST_F(SimpleTupleStreamTest, UnpinReadPage) {
write_batch->Reset();
}
+// Test that UnpinStream defer advancing the read page when all rows from the read page
+// are attached to a returned RowBatch but got not enough reservation.
+TEST_F(SimpleTupleStreamTest, DeferAdvancingReadPage) {
+ int num_rows = 1024;
+ int buffer_size = 4 * 1024;
+ // Only give 2 * buffer_size for the stream initial read and write page reservation.
+ Init(2 * buffer_size);
+
+ bool eos;
+ bool got_reservation;
+ Status status;
+ RowBatch* write_batch = CreateIntBatch(0, num_rows, false);
+
+ {
+ // Test unpinning a stream when the read page has been attached to the output batch
+ // and the output batch has NOT been reset.
+ BufferedTupleStream stream(
+ runtime_state_, int_desc_, &client_, buffer_size, buffer_size);
+ ASSERT_OK(stream.Init("SimpleTupleStreamTest::DeferAdvancingReadPage", true));
+ ASSERT_OK(stream.PrepareForReadWrite(true, &got_reservation));
+ ASSERT_TRUE(got_reservation);
+
+ // Add rows to stream.
+ for (int i = 0; i < write_batch->num_rows(); ++i) {
+ EXPECT_TRUE(stream.AddRow(write_batch->GetRow(i), &status));
+ ASSERT_OK(status);
+ }
+
+ // Read until the read page is attached to the output.
+ RowBatch read_batch(int_desc_, num_rows, &tracker_);
+ ASSERT_OK(stream.GetNext(&read_batch, &eos));
+ // If GetNext did hit the capacity of the RowBatch, then the read page should have
+ // been attached to read_batch.
+ ASSERT_TRUE(read_batch.num_rows() < num_rows);
+ ASSERT_TRUE(!eos);
+
+ // We continue adding rows into the stream without releasing the read_batch. We expect
+ // that reservation limit will be hit and stream will need to be unpinned. We also
+ // expect that, after unpinning the stream, subsequent AddRow is always successful
+ // even if we're not immediately releasing the read_batch. We insert write_batch twice
+ // to ensure that we're inserting both in pinned and unpinned mode.
+ ASSERT_TRUE(stream.is_pinned());
+ for (int j = 0; j < 2; ++j) {
+ for (int i = 0; i < write_batch->num_rows(); ++i) {
+ bool succeed = stream.AddRow(write_batch->GetRow(i), &status);
+ ASSERT_TRUE(succeed || stream.is_pinned());
+ if (!succeed) {
+ // Unpin the stream.
+ status = stream.UnpinStream(BufferedTupleStream::UNPIN_ALL_EXCEPT_CURRENT);
+ ASSERT_OK(status);
+ ASSERT_FALSE(stream.is_pinned());
+ ASSERT_EQ(stream.bytes_unpinned(), 0);
+ ASSERT_EQ(stream.pages_.size(), 2);
+ ASSERT_EQ(stream.num_pages_, 2);
+ // Retry inserting this row by decreasing the index.
+ // After stream get into unpinned mode, further inserts should be successful,
+ // even if we're not immediately cleaning up the read_batch.
+ // Stream should be able to unpin the previous write page to reclaim some
+ // memory reservation to allocate new write page.
+ --i;
+ }
+ }
+ }
+ ASSERT_FALSE(stream.is_pinned());
+ stream.Close(nullptr, RowBatch::FlushMode::NO_FLUSH_RESOURCES);
+ read_batch.Reset();
+ }
+ write_batch->Reset();
+}
+
// Test writing to a stream (AddRow and UnpinStream), even though attached pages have not
// been released yet.
TEST_F(SimpleTupleStreamTest, WriteAfterReadAttached) {
diff --git a/be/src/runtime/buffered-tuple-stream.cc b/be/src/runtime/buffered-tuple-stream.cc
index a4b76a2..01edc84 100644
--- a/be/src/runtime/buffered-tuple-stream.cc
+++ b/be/src/runtime/buffered-tuple-stream.cc
@@ -716,14 +716,41 @@ Status BufferedTupleStream::UnpinStream(UnpinMode mode) {
if (pinned_) {
CHECK_CONSISTENCY_FULL(read_it_);
+ bool defer_advancing_read_page = false;
if (&*read_it_.read_page_ != write_page_ && read_it_.read_page_ != pages_.end()
&& read_it_.read_page_rows_returned_ == read_it_.read_page_->num_rows) {
- RETURN_IF_ERROR(NextReadPage(&read_it_));
+ if (read_it_.read_page_->attached_to_output_batch) {
+ if (num_pages_ <= 2) {
+ // NextReadPage will attempt to save default_page_len_ into write reservation if
+ // the stream ended up with only 1 read/write page after advancing the read
+ // page. This can potentially lead to negative unused reservation if the reader
+ // has not freed the row batch where the read page buffer is attached to. We
+ // defer advancing the read page until the next GetNext() call by the reader
+ // (see IMPALA-10584).
+ defer_advancing_read_page = true;
+ }
+ }
+
+ if (!defer_advancing_read_page) {
+ RETURN_IF_ERROR(NextReadPage(&read_it_));
+ }
}
// If the stream was pinned, there may be some remaining pinned pages that should
// be unpinned at this point.
- for (Page& page : pages_) UnpinPageIfNeeded(&page, false);
+ DCHECK_EQ(bytes_unpinned_, 0);
+ std::list<Page>::iterator it = pages_.begin();
+ if (defer_advancing_read_page) {
+ // We skip advancing the read page earlier, so the first page must be a read page
+ // and attached_to_output_batch is true. We should keep the first page pinned. The
+ // next GetNext() call is the one who will be responsible to unpin the first page.
+ DCHECK(read_it_.read_page_ == pages_.begin());
+ ++it;
+ }
+ while (it != pages_.end()) {
+ UnpinPageIfNeeded(&(*it), false);
+ ++it;
+ }
// Check to see if we need to save some of the reservation we freed up.
if (!NeedWriteReservation(true) && NeedWriteReservation(false)) {
diff --git a/be/src/runtime/buffered-tuple-stream.h b/be/src/runtime/buffered-tuple-stream.h
index 4f76cba..bba6479 100644
--- a/be/src/runtime/buffered-tuple-stream.h
+++ b/be/src/runtime/buffered-tuple-stream.h
@@ -561,6 +561,7 @@ class BufferedTupleStream {
private:
DISALLOW_COPY_AND_ASSIGN(BufferedTupleStream);
friend class SimpleTupleStreamTest_ShortDebugString_Test;
+ friend class SimpleTupleStreamTest_DeferAdvancingReadPage_Test;
/// Runtime state instance used to check for cancellation. Not owned.
RuntimeState* const state_;
diff --git a/tests/query_test/test_scratch_limit.py b/tests/query_test/test_scratch_limit.py
index 07f34da..8779cd3 100644
--- a/tests/query_test/test_scratch_limit.py
+++ b/tests/query_test/test_scratch_limit.py
@@ -71,7 +71,6 @@ class TestScratchLimit(ImpalaTestSuite):
exec_option = vector.get_value('exec_option')
exec_option['buffer_pool_limit'] = self.buffer_pool_limit
exec_option['scratch_limit'] = '500m'
- exec_option['spool_query_results'] = '0'
self.execute_query_expect_success(self.client, self.spilling_sort_query, exec_option)
def test_with_low_scratch_limit(self, vector):
@@ -82,7 +81,6 @@ class TestScratchLimit(ImpalaTestSuite):
exec_option = vector.get_value('exec_option')
exec_option['buffer_pool_limit'] = self.buffer_pool_limit
exec_option['scratch_limit'] = '24m'
- exec_option['spool_query_results'] = '0'
expected_error = 'Scratch space limit of %s bytes exceeded'
scratch_limit_in_bytes = 24 * 1024 * 1024
try:
@@ -99,7 +97,6 @@ class TestScratchLimit(ImpalaTestSuite):
exec_option = vector.get_value('exec_option')
exec_option['buffer_pool_limit'] = self.buffer_pool_limit
exec_option['scratch_limit'] = '0'
- exec_option['spool_query_results'] = '0'
for query in self.spilling_queries:
self.execute_query_expect_failure(query, exec_option)
@@ -110,7 +107,6 @@ class TestScratchLimit(ImpalaTestSuite):
exec_option = vector.get_value('exec_option')
exec_option['buffer_pool_limit'] = self.buffer_pool_limit
exec_option['scratch_limit'] = '-1'
- exec_option['spool_query_results'] = '0'
self.execute_query_expect_success(self.client, self.spilling_sort_query, exec_option)
def test_without_specifying_scratch_limit(self, vector):
@@ -119,7 +115,6 @@ class TestScratchLimit(ImpalaTestSuite):
"""
exec_option = vector.get_value('exec_option')
exec_option['buffer_pool_limit'] = self.buffer_pool_limit
- exec_option['spool_query_results'] = '0'
self.execute_query_expect_success(self.client, self.spilling_sort_query, exec_option)
def test_with_zero_scratch_limit_no_memory_limit(self, vector):
@@ -129,7 +124,6 @@ class TestScratchLimit(ImpalaTestSuite):
"""
exec_option = vector.get_value('exec_option')
exec_option['scratch_limit'] = '0'
- exec_option['spool_query_results'] = '0'
for query in self.spilling_queries:
self.execute_query_expect_success(self.client, query, exec_option)