Hive splits and multithreading

[saifmasood]

My understanding is that we can provide splits to velox and each split will be run in parallel (limited by the maxDrivers configuration).

In the following code snippet, I've created 4 hive splits and added them to the task. However, Velox treats them as one split and executes the task, even though it can be seen from the output that 4 splits were created. As a result, there is no parallelism. Could you please help me understand if I'm doing something wrong here or if there is a gap in my understanding?

void MLFunctionsTest::test_multithreading() { 
  int input_size = 1000;
  int output_size = 500;
  int num_samples = 600;
  int size = output_size * input_size;
  
  auto weights = maker.flatVector<float>(size);

  for(int i=0; i < size; i++){
    weights->set(i, i*10);
  } 
  // register Vector Function
  exec::registerVectorFunction(
    "mat_mul",
    MatrixMultiply::signatures(),
    std::make_unique<MatrixMultiply>(weights->values()->asMutable<float>(), input_size, output_size)
  );

  // Create input
  std::vector<std::vector<float>> featureVectors;
  for(int i=0; i < num_samples; i++){ 
    std::vector<float> featureVector;
    for(int j=0; j < input_size; j++){
      featureVector.push_back(i*j);
    }
    featureVectors.push_back(featureVector);
  }
  auto featureArrayVector = maker.arrayVector<float>(featureVectors, REAL());
  auto inputRowVector = maker.rowVector({"x"}, {featureArrayVector});
  
  // Create Plan
  auto planNodeIdGenerator = std::make_shared<core::PlanNodeIdGenerator>();
  core::PlanNodeId p0;
  auto plan0 = exec::test::PlanBuilder(planNodeIdGenerator, childPool.get())
                  .tableScan(asRowType(inputRowVector->type()))
                  .capturePlanNodeId(p0)
		              .project({"mat_mul(x)"})
                  .planFragment();

  // Create task
  auto task = exec::Task::create("0", plan0 , 0, queryCtx_, 
        [](RowVectorPtr /*unused*/, ContinueFuture* /*unused*/) {
          return exec::BlockingReason::kNotBlocked;
  });

  // Create 4 hive splits and add them to task
  auto file = TempFilePath::create();
  writeToFile(file->path, {inputRowVector});
  auto hiveSplits = makeHiveConnectorSplits(file->path, 4, dwio::common::FileFormat::DWRF);
  std::cout << "Hive splits:" << std::endl;
  for(auto& split : hiveSplits) {
    std::cout << split->toString() << std::endl;
    task->addSplit(p0, exec::Split(std::move(split)));
  }
  task->noMoreSplits(p0);
  std::cout << std::endl;
  // Start task with 2 as maximum drivers and wait for execution to finish
  task->start(task, 2);
  std::chrono::steady_clock::time_point begin = std::chrono::steady_clock::now();
  waitForFinishedDrivers(task);
  std::chrono::steady_clock::time_point end = std::chrono::steady_clock::now();
  std::cout << "Total time (sec) = " <<  (std::chrono::duration_cast<std::chrono::microseconds>(end - begin).count()) /1000000.0 << std::endl;
}

MatrixMultiply signature

static std::vector<std::shared_ptr<exec::FunctionSignature>> signatures() {
        return {exec::FunctionSignatureBuilder()
                     .returnType("array(REAL)")
                     .argumentType("array(REAL)")
                     .build()};
}

Output

I0706 17:14:06.775618 1030378 Compression.cpp:461] Initialized zstd compressor with compression level 7
Hive splits:
Hive: file:/tmp/velox_test_CGLiSx 0 - 400499
Hive: file:/tmp/velox_test_CGLiSx 400499 - 400499
Hive: file:/tmp/velox_test_CGLiSx 800998 - 400499
Hive: file:/tmp/velox_test_CGLiSx 1201497 - 400499

Matrix shapes Matmul
Matrix shape: 600 x 1000
Matrix shape: 1000 x 500
Time for Matrix multiply (sec) = 0.885307
Total time (sec) = 1.00069

[xiaoxmeng]

@saifmasood I think Velox does parallelize the table scan operation at split level. The split added by Task::addSplit() used in test actually put the splits in a shared groupSplitsStores in the task which are indexed by table scan plan node id so all the table scan operators could access the added splits. Each table scan operator fetches and processes one split at time (table scan operator's getOutput method calls Task::getSplitOrFuture() to do that). I am not sure how you find out Velox processes all the four as one? There might be some time related race condition that one table scan operator run very fast and process all the four even before the second one starts but you can experiment more splits.

[xiaoxmeng]

Note that the table scan works at one group at a time and if the file is small and only contains one row group, then only the first split has the actual data to process and all the rest are empty. The split doesn't need to start or end at the row group boundary and the table scan operator will continue process beyond the end of split until finish one row group. Correspondingly, the table scan operator will skip processing the row group if the split starts in the middle of a row group.

[saifmasood]

Thanks for the pointers @xiaoxmeng . Yes, the issue was that even though 4 splits were being created, the first one had the entire data and the rest were empty. Tweaking configs hive.exec.orc.row.index.stride and hive.exec.orc.stripe.size helped in this case.