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