Test for advancing the read/write page with resource flushing.
| 949 | |
| 950 | // Test for advancing the read/write page with resource flushing. |
| 951 | void SimpleTupleStreamTest::TestFlushResourcesReadWrite( |
| 952 | bool pin_stream, bool attach_on_read) { |
| 953 | // Use smaller buffers so that the explicit FLUSH_RESOURCES flag is required to |
| 954 | // make the batch at capacity. |
| 955 | const int BUFFER_SIZE = 512; |
| 956 | const int BATCH_SIZE = 100; |
| 957 | // For unpinned streams, we should be able to iterate with only two buffers. |
| 958 | const int MAX_PINNED_PAGES = pin_stream ? 1000 : 2; |
| 959 | Init(MAX_PINNED_PAGES * BUFFER_SIZE); |
| 960 | |
| 961 | BufferedTupleStream stream( |
| 962 | runtime_state_, int_desc_, &client_, BUFFER_SIZE, BUFFER_SIZE); |
| 963 | ASSERT_OK( |
| 964 | stream.Init("SimpleTupleStreamTest::TestFlushResourcesReadWrite", pin_stream)); |
| 965 | bool got_reservation; |
| 966 | ASSERT_OK(stream.PrepareForReadWrite(attach_on_read, &got_reservation)); |
| 967 | ASSERT_TRUE(got_reservation); |
| 968 | int num_buffers_attached = 0; |
| 969 | /// Read over the page in different increments. |
| 970 | for (int append_batch_size : {1, 10, 100, 1000}) { |
| 971 | AppendToReadWriteStream( |
| 972 | append_batch_size, BUFFER_SIZE, &num_buffers_attached, &stream); |
| 973 | } |
| 974 | |
| 975 | if (attach_on_read) { |
| 976 | EXPECT_EQ(stream.byte_size() / BUFFER_SIZE - 1, num_buffers_attached) |
| 977 | << "All buffers except the current write page should have been attached"; |
| 978 | } else { |
| 979 | EXPECT_EQ(0, num_buffers_attached); |
| 980 | } |
| 981 | |
| 982 | RowBatch* final_out_batch = pool_.Add(new RowBatch(int_desc_, BATCH_SIZE, &tracker_)); |
| 983 | stream.Close(final_out_batch, RowBatch::FlushMode::FLUSH_RESOURCES); |
| 984 | final_out_batch->Reset(); |
| 985 | } |
| 986 | |
| 987 | void SimpleTupleStreamTest::AppendToReadWriteStream(int64_t append_batch_size, |
| 988 | int64_t buffer_size, int* num_buffers_attached, BufferedTupleStream* stream) { |