| 559 | } |
| 560 | |
| 561 | void verifySortedSpillData( |
| 562 | SpillPartition* spillPartition, |
| 563 | int32_t outputBatchSize = 0, |
| 564 | bool spillUringEnabled = false) { |
| 565 | ASSERT_EQ(numPartitions_, 1); |
| 566 | ASSERT_TRUE(spiller_->isSpilled(0)); |
| 567 | |
| 568 | // We make a merge reader that merges the spill files and the rows that |
| 569 | // are still in the RowContainer. |
| 570 | auto merge = spillPartition->createOrderedReader(pool(), spillUringEnabled); |
| 571 | ASSERT_TRUE(merge != nullptr); |
| 572 | ASSERT_TRUE( |
| 573 | spillPartition->createOrderedReader(pool(), spillUringEnabled) == |
| 574 | nullptr); |
| 575 | |
| 576 | // We read the spilled data back and check that it matches the sorted |
| 577 | // order of the partition. |
| 578 | auto& indices = partitions_[0]; |
| 579 | if (outputBatchSize == 0) { |
| 580 | for (auto i = 0; i < indices.size(); ++i) { |
| 581 | auto stream = merge->next(); |
| 582 | if (!stream) { |
| 583 | FAIL() << "Stream ends after " << i << " entries"; |
| 584 | break; |
| 585 | } |
| 586 | ASSERT_TRUE(rowVector_->equalValueAt( |
| 587 | &stream->current(), indices[i], stream->currentIndex())); |
| 588 | stream->pop(); |
| 589 | } |
| 590 | } else { |
| 591 | int nextBatchSize = std::min<int>(indices.size(), outputBatchSize); |
| 592 | auto outputVector = BaseVector::create<RowVector>( |
| 593 | rowVector_->type(), nextBatchSize, pool_.get()); |
| 594 | resizeVector(*outputVector, nextBatchSize); |
| 595 | |
| 596 | int i = 0; |
| 597 | int outputRow = 0; |
| 598 | int outputSize = 0; |
| 599 | std::vector<const RowVector*> sourceVectors(outputBatchSize); |
| 600 | std::vector<vector_size_t> sourceIndices(outputBatchSize); |
| 601 | for (;;) { |
| 602 | auto stream = merge->next(); |
| 603 | if (stream == nullptr) { |
| 604 | for (int j = 0; j < outputVector->size(); ++j, ++i) { |
| 605 | ASSERT_TRUE( |
| 606 | rowVector_->equalValueAt(outputVector.get(), indices[i], j)) |
| 607 | << j << ", " << i; |
| 608 | } |
| 609 | ASSERT_EQ(i, indices.size()); |
| 610 | break; |
| 611 | } |
| 612 | sourceVectors[outputSize] = &stream->current(); |
| 613 | bool isEndOfBatch = false; |
| 614 | sourceIndices[outputSize] = stream->currentIndex(&isEndOfBatch); |
| 615 | ++outputSize; |
| 616 | if (isEndOfBatch) { |
| 617 | // The stream is at end of input batch. Need to copy out the rows |
| 618 | // before fetching next batch in 'pop'. |
nothing calls this directly
no test coverage detected