Operator addInput time is aggregated in source operator statistics

[LeonidChistov]

Bug description

Execution statistics (time, count, etc) of each operator addInput method call is currently associated and collected as part of source operator statistics.

In the current code

velox/velox/exec/Driver.cpp

Line 618 in 48e3a4d

withDeltaCpuWallTimer(op, &OperatorStats::addInputTiming, [&]() {

we see following logic

withDeltaCpuWallTimer(op, &OperatorStats::addInputTiming, [&]() {
               ...
                CALL_OPERATOR(
                    nextOp->addInput(intermediateResult),
                    nextOp,
                    curOperatorId_ + 1,
                    kOpMethodAddInput);
              });

which causes nextOp->addInput statistics to be aggregated to input timing of op.

It looks like this logic was introduced in #10906 and it was different before:

 auto timer = createDeltaCpuWallTimer(
                  [nextOp, this](const CpuWallTiming& elapsedTime) {
                    auto elapsedSelfTime =
                        processLazyTiming(*nextOp, elapsedTime);
                    nextOp->stats().wlock()->addInputTiming.add(
                        elapsedSelfTime);
                  });

which probably means that it was just a typo in the Pull Request.

Current logic causes following statistics effect on the UI level, for example

Here we have large CPU time attributed to LocalExchange although it actually relates to HashAggregate::addInput run time and no CPU time for HashBuild since it is also "moved down" in the pipeline.

System information

/

Relevant logs

No response