Method testSortedSpill

bolt/exec/tests/SpillerTest.cpp:194–312

  }

  void testSortedSpill(
      int numDuplicates,
      int32_t outputBatchSize = 0,
      bool ascending = true,
      bool makeError = false) {
    SCOPED_TRACE(fmt::format(
        "spillType: {} numDuplicates: {} outputBatchSize: {} ascending: {} makeError: {}",
        Spiller::typeName(type_),
        numDuplicates,
        outputBatchSize,
        ascending,
        makeError));
    constexpr int32_t kNumRows = 5'000;
    const auto prevGStats = common::globalSpillStats();

    setupSpillData(
        rowType_, numKeys_, kNumRows, numDuplicates, [&](RowVectorPtr rows) {
          // Set ordinal so that the sorted order is unambiguous.
          setSequentialValue(rows, 5);
        });
    sortSpillData(ascending);

    setupSpiller(2'000'000, 0, makeError);

    // We spill spillPct% of the data in 10% increments.
    runSpill(makeError);
    if (makeError) {
      return;
    }
    // Verify the spilled file exist on file system.
    auto stats = spiller_->stats();
    const auto numSpilledFiles = stats.spilledFiles;
    if (type_ == Spiller::Type::kAggregateOutput) {
      ASSERT_EQ(numSpilledFiles, 1);
    } else {
      ASSERT_GT(numSpilledFiles, 0);
    }
    const auto spilledFileSet = spiller_->state().testingSpilledFilePaths();
    ASSERT_EQ(spilledFileSet.size(), numSpilledFiles);

    uint64_t totalSpilledBytes{0};
    for (auto spilledFile : spilledFileSet) {
      auto readFile = fs_->openFileForRead(spilledFile);
      ASSERT_NE(readFile.get(), nullptr);
      totalSpilledBytes += readFile->size();
    }
    ASSERT_TRUE(spiller_->isAnySpilled());
    ASSERT_TRUE(spiller_->isAllSpilled());
    ASSERT_FALSE(spiller_->finalized());
    BOLT_ASSERT_THROW(spiller_->spill(0, nullptr), "Unexpected spiller type");
    BOLT_ASSERT_THROW(
        spiller_->setPartitionsSpilled({}), "Unexpected spiller type");
    auto spillPartition = spiller_->finishSpill();
    ASSERT_TRUE(spiller_->finalized());
    ASSERT_EQ(rowContainer_->numRows(), 0);
    ASSERT_EQ(numPartitions_, spiller_->stats().spilledPartitions);
    ASSERT_EQ(numPartitions_, spiller_->state().spilledPartitionSet().size());
    ASSERT_EQ(numSpilledFiles, spiller_->stats().spilledFiles);