Method AppendToReadWriteStream

be/src/runtime/buffered-tuple-stream-test.cc:987–1031

}

void SimpleTupleStreamTest::AppendToReadWriteStream(int64_t append_batch_size,
    int64_t buffer_size, int* num_buffers_attached, BufferedTupleStream* stream) {
  RowBatch* in_batch = CreateIntBatch(0, BATCH_SIZE, false);

  /// Accumulate row batches until we see a flush. The contents of the batches should
  /// remain valid until reset or delete trailing batches.
  vector<unique_ptr<RowBatch>> out_batches;
  // The start row index of each batch in 'out_batches'.
  vector<int64_t> out_batch_start_indices;
  // Iterate over at least 10 pages.
  int64_t start_byte_size = stream->byte_size();
  while (stream->byte_size() - start_byte_size < 10 * buffer_size) {
    Status status;
    for (int i = 0; i < append_batch_size; ++i) {
      bool ret = stream->AddRow(
          in_batch->GetRow(stream->num_rows() % in_batch->num_rows()), &status);
      EXPECT_TRUE(ret);
      ASSERT_OK(status);
    }
    int64_t rows_read = 0;
    bool eos;
    while (rows_read < append_batch_size) {
      out_batches.emplace_back(new RowBatch(int_desc_, BATCH_SIZE, &tracker_));
      out_batch_start_indices.push_back(stream->rows_returned());
      ASSERT_OK(stream->GetNext(out_batches.back().get(), &eos));
      // Verify the contents of all valid batches to make sure that they haven't become
      // invalid.
      LOG(INFO) << "Verifying " << out_batches.size() << " batches";
      for (int i = 0; i < out_batches.size(); ++i) {
        VerifyReadWriteBatch(in_batch, out_batches[i].get(), out_batch_start_indices[i]);
      }
      *num_buffers_attached += out_batches.back()->num_buffers();
      rows_read += out_batches.back()->num_rows();
      EXPECT_EQ(rows_read == append_batch_size, eos);
      if (out_batches.back().get()->flush_mode()
          == RowBatch::FlushMode::FLUSH_RESOURCES) {
        out_batches.clear();
        out_batch_start_indices.clear();
      }
    }
    EXPECT_EQ(append_batch_size, rows_read);
    EXPECT_EQ(true, eos);
  }
  in_batch->Reset();
}

void SimpleTupleStreamTest::VerifyReadWriteBatch(
    RowBatch* in_batch, RowBatch* out_batch, int64_t start_index) {