Intermediate changes

commit_hash:e7fe74e96542f1fb4565640d02a416595ad3d438

25 files changed, 246 insertions, 55 deletions

diff --git a/yql/essentials/tools/purebench/purebench.cpp b/yql/essentials/tools/purebench/purebench.cpp
index 87b70376d1e..7724cb73594 100644
--- a/yql/essentials/tools/purebench/purebench.cpp
+++ b/yql/essentials/tools/purebench/purebench.cpp
@@ -176,6 +176,7 @@ public:
                 stream->Write(&len, sizeof(len));
                 stream->Write(buf.Data(), len);
             }
+            Worker_->CheckState(true);
         }
     }
 
@@ -263,6 +264,103 @@ struct TOutputSpecTraits<TPrintOutputSpec> {
     }
 };
 
+template <bool SupportsBlocks>
+struct TNopOutputSpec: public TOutputSpecBase {
+    explicit TNopOutputSpec(NYT::TNode schema)
+        : Schema(std::move(schema))
+    {
+    }
+
+    const NYT::TNode& GetSchema() const final {
+        return Schema;
+    }
+
+    bool AcceptsBlocks() const override {
+        return SupportsBlocks;
+    }
+
+    const NYT::TNode Schema;
+};
+
+class TNopScalarOutputHandle final: public TStreamOutputHandle {
+public:
+    explicit TNopScalarOutputHandle(TWorkerHolder<IPullListWorker> worker)
+        : Worker_(std::move(worker))
+    {
+    }
+
+    NKikimr::NMiniKQL::TType* GetOutputType() const final {
+        return const_cast<NKikimr::NMiniKQL::TType*>(Worker_->GetOutputType());
+    }
+
+    void Run(IOutputStream* stream) final {
+        Y_UNUSED(stream);
+        Y_ENSURE(
+            Worker_->GetOutputType()->IsStruct(),
+            "Run(IOutputStream*) cannot be used with multi-output programs");
+
+        TBindTerminator bind(Worker_->GetGraph().GetTerminator());
+
+        with_lock (Worker_->GetScopedAlloc()) {
+            const auto outputIterator = Worker_->GetOutputIterator();
+
+            TUnboxedValue value;
+            while (outputIterator.Next(value)) {
+            }
+            Worker_->CheckState(true);
+        }
+    }
+
+private:
+    TWorkerHolder<IPullListWorker> Worker_;
+};
+
+class TNopBlockOutputHandle final: public IStream<arrow::compute::ExecBatch*> {
+public:
+    explicit TNopBlockOutputHandle(TWorkerHolder<IPullListWorker> worker)
+        : Worker_(std::move(worker))
+    {
+    }
+
+    arrow::compute::ExecBatch* Fetch() override {
+        TBindTerminator bind(Worker_->GetGraph().GetTerminator());
+
+        with_lock (Worker_->GetScopedAlloc()) {
+            const auto outputIterator = Worker_->GetOutputIterator();
+
+            TUnboxedValue value;
+            if (outputIterator.Next(value)) {
+                return &ExecBatchStub_;
+            }
+            Worker_->CheckState(true);
+            return nullptr;
+        }
+    }
+
+private:
+    TWorkerHolder<IPullListWorker> Worker_;
+    arrow::compute::ExecBatch ExecBatchStub_;
+};
+
+template <bool SupportsBlocks>
+struct TOutputSpecTraits<TNopOutputSpec<SupportsBlocks>> {
+    static const constexpr bool IsPartial = false;
+
+    static const constexpr bool SupportPullStreamMode = false;
+    static const constexpr bool SupportPullListMode = true;
+    static const constexpr bool SupportPushStreamMode = false;
+
+    using TPullListReturnType = std::conditional_t<SupportsBlocks, THolder<TNopBlockOutputHandle>, THolder<TNopScalarOutputHandle>>;
+
+    static TPullListReturnType ConvertPullListWorkerToOutputType(const TNopOutputSpec<SupportsBlocks>&, TWorkerHolder<IPullListWorker> worker) {
+        if constexpr (SupportsBlocks) {
+            return MakeHolder<TNopBlockOutputHandle>(std::move(worker));
+        } else {
+            return MakeHolder<TNopScalarOutputHandle>(std::move(worker));
+        }
+    }
+};
+
 TStringStream MakeGenInput(ui64 count) {
     TStringStream stream;
     TScopedAlloc alloc(__LOCATION__);
@@ -288,16 +386,13 @@ TStringStream MakeGenInput(ui64 count) {
     return stream;
 }
 
-template <typename TInputSpec, typename TOutputSpec>
-using TRunCallable = std::function<void(const THolder<TPullListProgram<TInputSpec, TOutputSpec>>&)>;
-
-template <typename TOutputSpec>
+template <typename TOutputSpec, typename TRunProgram>
 NYT::TNode RunGenSql(
     const IProgramFactoryPtr factory,
     const TVector<NYT::TNode>& inputSchema,
     const TString& sql,
     ETranslationMode isPg,
-    TRunCallable<TPickleInputSpec, TOutputSpec> runCallable) {
+    const TRunProgram& runCallable) {
     auto inputSpec = TPickleInputSpec(inputSchema);
     auto outputSpec = TOutputSpec({NYT::TNode::CreateEntity()});
     auto program = factory->MakePullListProgram(inputSpec, outputSpec, sql, isPg);
@@ -329,41 +424,69 @@ void ShowResults(
     Cerr << "\n";
 }
 
-template <typename TInputSpec, typename TOutputSpec>
+double Score(TVector<TDuration>&& times) {
+    Sort(times);
+    times.erase(times.end() - times.size() / 3, times.end());
+
+    double sum = std::transform_reduce(times.cbegin(), times.cend(),
+                                       .0, std::plus{}, [](auto t) { return std::log(t.MicroSeconds()); });
+
+    return std::exp(sum / times.size());
+}
+
+template <typename TInputSpec, typename TOutputSpec, typename TNopOutputSpec, typename TRunProgram>
 double RunBenchmarks(
-    const IProgramFactoryPtr factory,
+    const IProgramFactoryPtr testFactory,
+    const IProgramFactoryPtr benchFactory,
     const TVector<NYT::TNode>& inputSchema,
     const TString& sql,
     ETranslationMode isPg,
-    ui32 repeats,
-    TRunCallable<TInputSpec, TOutputSpec> runCallable) {
+    ui32 benchmarkRuns,
+    ui32 calibrationRuns,
+    TDuration benchmarkTime,
+    TDuration calibrationTime,
+    const TRunProgram& runCallable) {
     auto inputSpec = TInputSpec(inputSchema);
     auto outputSpec = TOutputSpec({NYT::TNode::CreateEntity()});
-    auto program = factory->MakePullListProgram(inputSpec, outputSpec, sql, isPg);
+    auto nopSpec = TNopOutputSpec({NYT::TNode::CreateEntity()});
 
     Cerr << "Dry run of test sql...\n";
 
-    runCallable(program);
+    auto testProgram = testFactory->MakePullListProgram(inputSpec, outputSpec, sql, isPg);
+
+    runCallable(testProgram);
 
     Cerr << "Run benchmark...\n";
 
-    TVector<TDuration> times;
-    TSimpleTimer allTimer;
-    for (ui32 i = 0; i < repeats; ++i) {
+    auto benchProgram = benchFactory->MakePullListProgram(inputSpec, nopSpec, sql, isPg);
+
+    ui32 benchmarks = 0;
+    TVector<TDuration> benchTimes;
+    TSimpleTimer benchTimer;
+    while (++benchmarks < benchmarkRuns || benchTimer.Get() < benchmarkTime) {
         TSimpleTimer timer;
-        runCallable(program);
-        times.push_back(timer.Get());
+        runCallable(benchProgram);
+        benchTimes.push_back(timer.Get());
     }
 
-    Cout << "Elapsed: " << allTimer.Get() << "\n";
+    Cout << "Benchmark completed: " << benchmarks << " iterations for " << benchTimer.Get() << "\n";
 
-    Sort(times);
-    times.erase(times.end() - times.size() / 3, times.end());
+    Cerr << "Score calibration...\n";
 
-    double sum = std::transform_reduce(times.cbegin(), times.cend(),
-                                       .0, std::plus{}, [](auto t) { return std::log(t.MicroSeconds()); });
+    auto calibrationProgram = benchFactory->MakePullListProgram(inputSpec, nopSpec, "SELECT * FROM Input", isPg);
 
-    return std::exp(sum / times.size());
+    ui32 calibrations = 0;
+    TVector<TDuration> calibrationTimes;
+    TSimpleTimer calibrationTimer;
+    while (++calibrations < calibrationRuns || calibrationTimer.Get() < calibrationTime) {
+        TSimpleTimer timer;
+        runCallable(calibrationProgram);
+        calibrationTimes.push_back(timer.Get());
+    }
+
+    Cerr << "Calibration completed: " << calibrations << " iterations for " << calibrationTimer.Get() << "\n";
+
+    return Score(std::move(calibrationTimes)) / Score(std::move(benchTimes)) * 100;
 }
 
 int Main(int argc, const char** argv)
@@ -372,7 +495,11 @@ int Main(int argc, const char** argv)
     using namespace NLastGetopt;
     TOpts opts = TOpts::Default();
     ui64 count;
-    ui32 repeats;
+    ui32 benchmarkRuns;
+    ui32 calibrationRuns;
+    TDuration benchmarkSeconds;
+    TDuration calibrationSeconds;
+    const auto secondsFromString = [](const TString& str) { return TDuration::Seconds(std::stoul(str)); };
     TString genSql, testSql;
     bool showResults;
     TString udfsDir;
@@ -386,7 +513,8 @@ int Main(int argc, const char** argv)
     opts.AddLongOption('c', "count", "count of input rows").StoreResult(&count).DefaultValue(1000000);
     opts.AddLongOption('g', "gen-sql", "SQL query to generate data").StoreResult(&genSql).DefaultValue("select index from Input");
     opts.AddLongOption('t', "test-sql", "SQL query to test").StoreResult(&testSql).DefaultValue("select count(*) as count from Input");
-    opts.AddLongOption('r', "repeats", "number of iterations").StoreResult(&repeats).DefaultValue(10);
+    opts.AddLongOption('r', "repeats", "number of iterations").StoreResult(&benchmarkRuns).DefaultValue(10);
+    opts.AddLongOption('R', "repeat-time", "total time running benchmark").StoreMappedResultT<TString, TDuration>(&benchmarkSeconds, secondsFromString).DefaultValue(1);
     opts.AddLongOption('w', "show-results", "show results of test SQL").StoreResult(&showResults).DefaultValue(true);
     opts.AddLongOption("pg", "use PG syntax for generate query").NoArgument();
     opts.AddLongOption("pt", "use PG syntax for test query").NoArgument();
@@ -394,6 +522,8 @@ int Main(int argc, const char** argv)
     opts.AddLongOption("llvm-settings", "LLVM settings").StoreResult(&LLVMSettings).DefaultValue("");
     opts.AddLongOption("print-expr", "print rebuild AST before execution").NoArgument();
     opts.AddLongOption("expr-file", "print AST to that file instead of stdout").StoreResult(&exprFile);
+    opts.AddLongOption("calibrate", "number of calibrating iterations").StoreResult(&calibrationRuns).DefaultValue(3);
+    opts.AddLongOption("calibrate-time", "number of calibrating iterations").StoreMappedResultT<TString, TDuration>(&calibrationSeconds, secondsFromString).DefaultValue(1);
     opts.AddLongOption("langver", "Set current language version").RequiredArgument("VER").Handler1T<TString>([&langVer](const TString& str) {
         if (str == "unknown") {
             langVer = UnknownLangVersion;
@@ -410,6 +540,8 @@ int Main(int argc, const char** argv)
     factoryOptions.SetBlockEngineSettings(blockEngineSettings);
     factoryOptions.SetLanguageVersion(langVer);
 
+    auto benchFactory = MakeProgramFactory(factoryOptions);
+
     IOutputStream* exprOut = nullptr;
     THolder<TFixedBufferFileOutput> exprFileHolder;
     if (res.Has("print-expr")) {
@@ -420,7 +552,7 @@ int Main(int argc, const char** argv)
     }
     factoryOptions.SetExprOutputStream(exprOut);
 
-    auto factory = MakeProgramFactory(factoryOptions);
+    auto testFactory = MakeProgramFactory(factoryOptions);
 
     NYT::TNode members{NYT::TNode::CreateList()};
     auto typeNode = NYT::TNode::CreateList()
@@ -440,12 +572,11 @@ int Main(int argc, const char** argv)
     ETranslationMode isPgGen = res.Has("pg") ? ETranslationMode::PG : ETranslationMode::SQL;
     ETranslationMode isPgTest = res.Has("pt") ? ETranslationMode::PG : ETranslationMode::SQL;
     double normalizedTime;
-    size_t inputBenchSize;
 
     if (blockEngineSettings == "disable") {
         TStringStream outputGenStream;
         auto outputGenSchema = RunGenSql<TPickleOutputSpec>(
-            factory, inputGenSchema, genSql, isPgGen,
+            testFactory, inputGenSchema, genSql, isPgGen,
             [&](const auto& program) {
                 auto handle = program->Apply(&inputGenStream);
                 handle->Run(&outputGenStream);
@@ -455,12 +586,12 @@ int Main(int argc, const char** argv)
         if (showResults) {
             auto inputResStream = TStringStream(outputGenStream);
             ShowResults<TPickleInputSpec>(
-                factory, {outputGenSchema}, testSql, isPgTest, &inputResStream);
+                benchFactory, {outputGenSchema}, testSql, isPgTest, &inputResStream);
         }
 
-        inputBenchSize = outputGenStream.Size();
-        normalizedTime = RunBenchmarks<TPickleInputSpec, TPickleOutputSpec>(
-            factory, {outputGenSchema}, testSql, isPgTest, repeats,
+        normalizedTime = RunBenchmarks<TPickleInputSpec, TPickleOutputSpec, TNopOutputSpec<false>>(
+            testFactory, benchFactory, {outputGenSchema}, testSql, isPgTest,
+            benchmarkRuns, calibrationRuns, benchmarkSeconds, calibrationSeconds,
             [&](const auto& program) {
                 auto inputBorrowed = TStringStream(outputGenStream);
                 auto handle = program->Apply(&inputBorrowed);
@@ -477,7 +608,7 @@ int Main(int argc, const char** argv)
         // allocator. Hence, the program (i.e. worker) object should be created
         // at the very beginning of the block, or at least prior to all the
         // temporary batch storages (mind outputGenStream below).
-        auto program = factory->MakePullListProgram(
+        auto program = testFactory->MakePullListProgram(
             inputGenSpec, outputGenSpec, genSql, isPgGen);
 
         auto handle = program->Apply(&inputGenStream);
@@ -500,12 +631,12 @@ int Main(int argc, const char** argv)
             auto inputResStreamHolder = StreamFromVector(outputGenStream);
             auto inputResStream = inputResStreamHolder.Get();
             ShowResults<TArrowInputSpec>(
-                factory, {outputGenSchema}, testSql, isPgTest, inputResStream);
+                benchFactory, {outputGenSchema}, testSql, isPgTest, inputResStream);
         }
 
-        inputBenchSize = outputGenSize;
-        normalizedTime = RunBenchmarks<TArrowInputSpec, TArrowOutputSpec>(
-            factory, {outputGenSchema}, testSql, isPgTest, repeats,
+        normalizedTime = RunBenchmarks<TArrowInputSpec, TArrowOutputSpec, TNopOutputSpec<true>>(
+            testFactory, benchFactory, {outputGenSchema}, testSql, isPgTest,
+            benchmarkRuns, calibrationRuns, benchmarkSeconds, calibrationSeconds,
             [&](const auto& program) {
                 auto handle = program->Apply(StreamFromVector(outputGenStream));
                 while (/* arrow::compute::ExecBatch* batch = */ handle->Fetch()) {
@@ -513,7 +644,7 @@ int Main(int argc, const char** argv)
             });
     }
 
-    Cout << "Bench score: " << Prec(inputBenchSize / normalizedTime, 4) << "\n";
+    Cout << "Bench score: " << Prec(normalizedTime, 4) << "\n";
 
     NLog::CleanupLogger();
     return 0;
diff --git a/yql/essentials/tools/purebench/test/canondata/test.test_purebench_inc_auto-blocks_/err b/yql/essentials/tools/purebench/test/canondata/test.test_purebench_inc_auto-blocks_/err
index e6e40db9eed..748d60280f8 100644
--- a/yql/essentials/tools/purebench/test/canondata/test.test_purebench_inc_auto-blocks_/err
+++ b/yql/essentials/tools/purebench/test/canondata/test.test_purebench_inc_auto-blocks_/err
@@ -2,3 +2,5 @@ Input data size: 12000000
 Generated data size: 8007392
 Dry run of test sql...
 Run benchmark...
+Score calibration...
+Calibration completed: <ITERATIONS> iterations for <DURATION>s
diff --git a/yql/essentials/tools/purebench/test/canondata/test.test_purebench_inc_auto-blocks_/out b/yql/essentials/tools/purebench/test/canondata/test.test_purebench_inc_auto-blocks_/out
index 903a599094e..fcbb4d15f39 100644
--- a/yql/essentials/tools/purebench/test/canondata/test.test_purebench_inc_auto-blocks_/out
+++ b/yql/essentials/tools/purebench/test/canondata/test.test_purebench_inc_auto-blocks_/out
@@ -14,5 +14,5 @@ After optimization:
   (return (Collect (NarrowMap (ToFlow (WideMap (FromFlow (ExpandMap (ToFlow (BlockSelf '0)) $3)) $4)) (lambda '($9 $10) (AsStruct '($2 $10) '('column0 $9))))))
 )))))
 )
-Elapsed: <DURATION>s
+Benchmark completed: <ITERATIONS> iterations for <DURATION>s
 Bench score: <SCORE>
diff --git a/yql/essentials/tools/purebench/test/canondata/test.test_purebench_inc_for_42_auto-blocks_/err b/yql/essentials/tools/purebench/test/canondata/test.test_purebench_inc_for_42_auto-blocks_/err
index e6e40db9eed..748d60280f8 100644
--- a/yql/essentials/tools/purebench/test/canondata/test.test_purebench_inc_for_42_auto-blocks_/err
+++ b/yql/essentials/tools/purebench/test/canondata/test.test_purebench_inc_for_42_auto-blocks_/err
@@ -2,3 +2,5 @@ Input data size: 12000000
 Generated data size: 8007392
 Dry run of test sql...
 Run benchmark...
+Score calibration...
+Calibration completed: <ITERATIONS> iterations for <DURATION>s
diff --git a/yql/essentials/tools/purebench/test/canondata/test.test_purebench_inc_for_42_auto-blocks_/out b/yql/essentials/tools/purebench/test/canondata/test.test_purebench_inc_for_42_auto-blocks_/out
index 7ea05ff84d3..af378f4c240 100644
--- a/yql/essentials/tools/purebench/test/canondata/test.test_purebench_inc_for_42_auto-blocks_/out
+++ b/yql/essentials/tools/purebench/test/canondata/test.test_purebench_inc_for_42_auto-blocks_/out
@@ -18,5 +18,5 @@ After optimization:
   (return (Collect (NarrowMap (ToFlow (WideMap (BlockCompress (WideMap (FromFlow (ExpandMap (ToFlow (BlockSelf '0)) $3)) $4) '1) $5)) (lambda '($13 $14) (AsStruct '($2 $14) '('column0 $13))))))
 )))))
 )
-Elapsed: <DURATION>s
+Benchmark completed: <ITERATIONS> iterations for <DURATION>s
 Bench score: <SCORE>
diff --git a/yql/essentials/tools/purebench/test/canondata/test.test_purebench_inc_for_42_no-blocks_/err b/yql/essentials/tools/purebench/test/canondata/test.test_purebench_inc_for_42_no-blocks_/err
index 3e6f6b1cb53..276990da9f6 100644
--- a/yql/essentials/tools/purebench/test/canondata/test.test_purebench_inc_for_42_no-blocks_/err
+++ b/yql/essentials/tools/purebench/test/canondata/test.test_purebench_inc_for_42_no-blocks_/err
@@ -2,3 +2,5 @@ Input data size: 12000000
 Generated data size: 12000000
 Dry run of test sql...
 Run benchmark...
+Score calibration...
+Calibration completed: <ITERATIONS> iterations for <DURATION>s
diff --git a/yql/essentials/tools/purebench/test/canondata/test.test_purebench_inc_for_42_no-blocks_/out b/yql/essentials/tools/purebench/test/canondata/test.test_purebench_inc_for_42_no-blocks_/out
index 6dca19e4ec9..941d90cad89 100644
--- a/yql/essentials/tools/purebench/test/canondata/test.test_purebench_inc_for_42_no-blocks_/out
+++ b/yql/essentials/tools/purebench/test/canondata/test.test_purebench_inc_for_42_no-blocks_/out
@@ -6,5 +6,5 @@ After optimization:
 (
 (return (Map (Filter (Self '0) (lambda '($1) (== (Member $1 'index) (Int32 '42)))) (lambda '($2) (AsStruct '('column0 (+ (Member $2 'index) (Int32 '1)))))))
 )
-Elapsed: <DURATION>s
+Benchmark completed: <ITERATIONS> iterations for <DURATION>s
 Bench score: <SCORE>
diff --git a/yql/essentials/tools/purebench/test/canondata/test.test_purebench_inc_for_42_use-blocks_/err b/yql/essentials/tools/purebench/test/canondata/test.test_purebench_inc_for_42_use-blocks_/err
index e6e40db9eed..748d60280f8 100644
--- a/yql/essentials/tools/purebench/test/canondata/test.test_purebench_inc_for_42_use-blocks_/err
+++ b/yql/essentials/tools/purebench/test/canondata/test.test_purebench_inc_for_42_use-blocks_/err
@@ -2,3 +2,5 @@ Input data size: 12000000
 Generated data size: 8007392
 Dry run of test sql...
 Run benchmark...
+Score calibration...
+Calibration completed: <ITERATIONS> iterations for <DURATION>s
diff --git a/yql/essentials/tools/purebench/test/canondata/test.test_purebench_inc_for_42_use-blocks_/out b/yql/essentials/tools/purebench/test/canondata/test.test_purebench_inc_for_42_use-blocks_/out
index 7ea05ff84d3..af378f4c240 100644
--- a/yql/essentials/tools/purebench/test/canondata/test.test_purebench_inc_for_42_use-blocks_/out
+++ b/yql/essentials/tools/purebench/test/canondata/test.test_purebench_inc_for_42_use-blocks_/out
@@ -18,5 +18,5 @@ After optimization:
   (return (Collect (NarrowMap (ToFlow (WideMap (BlockCompress (WideMap (FromFlow (ExpandMap (ToFlow (BlockSelf '0)) $3)) $4) '1) $5)) (lambda '($13 $14) (AsStruct '($2 $14) '('column0 $13))))))
 )))))
 )
-Elapsed: <DURATION>s
+Benchmark completed: <ITERATIONS> iterations for <DURATION>s
 Bench score: <SCORE>
diff --git a/yql/essentials/tools/purebench/test/canondata/test.test_purebench_inc_no-blocks_/err b/yql/essentials/tools/purebench/test/canondata/test.test_purebench_inc_no-blocks_/err
index 3e6f6b1cb53..276990da9f6 100644
--- a/yql/essentials/tools/purebench/test/canondata/test.test_purebench_inc_no-blocks_/err
+++ b/yql/essentials/tools/purebench/test/canondata/test.test_purebench_inc_no-blocks_/err
@@ -2,3 +2,5 @@ Input data size: 12000000
 Generated data size: 12000000
 Dry run of test sql...
 Run benchmark...
+Score calibration...
+Calibration completed: <ITERATIONS> iterations for <DURATION>s
diff --git a/yql/essentials/tools/purebench/test/canondata/test.test_purebench_inc_no-blocks_/out b/yql/essentials/tools/purebench/test/canondata/test.test_purebench_inc_no-blocks_/out
index b70381e56ee..66f07ce1534 100644
--- a/yql/essentials/tools/purebench/test/canondata/test.test_purebench_inc_no-blocks_/out
+++ b/yql/essentials/tools/purebench/test/canondata/test.test_purebench_inc_no-blocks_/out
@@ -6,5 +6,5 @@ After optimization:
 (
 (return (Map (Self '0) (lambda '($1) (AsStruct '('column0 (+ (Member $1 'index) (Int32 '1)))))))
 )
-Elapsed: <DURATION>s
+Benchmark completed: <ITERATIONS> iterations for <DURATION>s
 Bench score: <SCORE>
diff --git a/yql/essentials/tools/purebench/test/canondata/test.test_purebench_inc_use-blocks_/err b/yql/essentials/tools/purebench/test/canondata/test.test_purebench_inc_use-blocks_/err
index e6e40db9eed..748d60280f8 100644
--- a/yql/essentials/tools/purebench/test/canondata/test.test_purebench_inc_use-blocks_/err
+++ b/yql/essentials/tools/purebench/test/canondata/test.test_purebench_inc_use-blocks_/err
@@ -2,3 +2,5 @@ Input data size: 12000000
 Generated data size: 8007392
 Dry run of test sql...
 Run benchmark...
+Score calibration...
+Calibration completed: <ITERATIONS> iterations for <DURATION>s
diff --git a/yql/essentials/tools/purebench/test/canondata/test.test_purebench_inc_use-blocks_/out b/yql/essentials/tools/purebench/test/canondata/test.test_purebench_inc_use-blocks_/out
index 903a599094e..fcbb4d15f39 100644
--- a/yql/essentials/tools/purebench/test/canondata/test.test_purebench_inc_use-blocks_/out
+++ b/yql/essentials/tools/purebench/test/canondata/test.test_purebench_inc_use-blocks_/out
@@ -14,5 +14,5 @@ After optimization:
   (return (Collect (NarrowMap (ToFlow (WideMap (FromFlow (ExpandMap (ToFlow (BlockSelf '0)) $3)) $4)) (lambda '($9 $10) (AsStruct '($2 $10) '('column0 $9))))))
 )))))
 )
-Elapsed: <DURATION>s
+Benchmark completed: <ITERATIONS> iterations for <DURATION>s
 Bench score: <SCORE>
diff --git a/yql/essentials/tools/purebench/test/canondata/test.test_purebench_langver_2025.04_/err b/yql/essentials/tools/purebench/test/canondata/test.test_purebench_langver_2025.04_/err
index d05e21fa837..1b03e475650 100644
--- a/yql/essentials/tools/purebench/test/canondata/test.test_purebench_langver_2025.04_/err
+++ b/yql/essentials/tools/purebench/test/canondata/test.test_purebench_langver_2025.04_/err
@@ -21,3 +21,5 @@ Generated data size: 12000000
 }
 Dry run of test sql...
 Run benchmark...
+Score calibration...
+Calibration completed: <ITERATIONS> iterations for <DURATION>s
diff --git a/yql/essentials/tools/purebench/test/canondata/test.test_purebench_langver_2025.04_/out b/yql/essentials/tools/purebench/test/canondata/test.test_purebench_langver_2025.04_/out
index f02ba5273b7..8032834491e 100644
--- a/yql/essentials/tools/purebench/test/canondata/test.test_purebench_langver_2025.04_/out
+++ b/yql/essentials/tools/purebench/test/canondata/test.test_purebench_langver_2025.04_/out
@@ -1,2 +1,2 @@
-Elapsed: <DURATION>s
+Benchmark completed: <ITERATIONS> iterations for <DURATION>s
 Bench score: <SCORE>
diff --git a/yql/essentials/tools/purebench/test/canondata/test.test_purebench_langver_noarg_/err b/yql/essentials/tools/purebench/test/canondata/test.test_purebench_langver_noarg_/err
index e0f55319511..4b8b48cde06 100644
--- a/yql/essentials/tools/purebench/test/canondata/test.test_purebench_langver_noarg_/err
+++ b/yql/essentials/tools/purebench/test/canondata/test.test_purebench_langver_noarg_/err
@@ -21,3 +21,5 @@ Generated data size: 12000000
 }
 Dry run of test sql...
 Run benchmark...
+Score calibration...
+Calibration completed: <ITERATIONS> iterations for <DURATION>s
diff --git a/yql/essentials/tools/purebench/test/canondata/test.test_purebench_langver_noarg_/out b/yql/essentials/tools/purebench/test/canondata/test.test_purebench_langver_noarg_/out
index f02ba5273b7..8032834491e 100644
--- a/yql/essentials/tools/purebench/test/canondata/test.test_purebench_langver_noarg_/out
+++ b/yql/essentials/tools/purebench/test/canondata/test.test_purebench_langver_noarg_/out
@@ -1,2 +1,2 @@
-Elapsed: <DURATION>s
+Benchmark completed: <ITERATIONS> iterations for <DURATION>s
 Bench score: <SCORE>
diff --git a/yql/essentials/tools/purebench/test/canondata/test.test_purebench_langver_unknown_/err b/yql/essentials/tools/purebench/test/canondata/test.test_purebench_langver_unknown_/err
index a8f89783ad9..4d87c49ac80 100644
--- a/yql/essentials/tools/purebench/test/canondata/test.test_purebench_langver_unknown_/err
+++ b/yql/essentials/tools/purebench/test/canondata/test.test_purebench_langver_unknown_/err
@@ -21,3 +21,5 @@ Generated data size: 12000000
 }
 Dry run of test sql...
 Run benchmark...
+Score calibration...
+Calibration completed: <ITERATIONS> iterations for <DURATION>s
diff --git a/yql/essentials/tools/purebench/test/canondata/test.test_purebench_langver_unknown_/out b/yql/essentials/tools/purebench/test/canondata/test.test_purebench_langver_unknown_/out
index f02ba5273b7..8032834491e 100644
--- a/yql/essentials/tools/purebench/test/canondata/test.test_purebench_langver_unknown_/out
+++ b/yql/essentials/tools/purebench/test/canondata/test.test_purebench_langver_unknown_/out
@@ -1,2 +1,2 @@
-Elapsed: <DURATION>s
+Benchmark completed: <ITERATIONS> iterations for <DURATION>s
 Bench score: <SCORE>
diff --git a/yql/essentials/tools/purebench/test/canondata/test.test_purebench_smoke_blocks_/err b/yql/essentials/tools/purebench/test/canondata/test.test_purebench_smoke_blocks_/err
index 505bc9374a6..fc45ebc864d 100644
--- a/yql/essentials/tools/purebench/test/canondata/test.test_purebench_smoke_blocks_/err
+++ b/yql/essentials/tools/purebench/test/canondata/test.test_purebench_smoke_blocks_/err
@@ -21,3 +21,5 @@ Generated data size: 8007392
 }
 Dry run of test sql...
 Run benchmark...
+Score calibration...
+Calibration completed: <ITERATIONS> iterations for <DURATION>s
diff --git a/yql/essentials/tools/purebench/test/canondata/test.test_purebench_smoke_blocks_/out b/yql/essentials/tools/purebench/test/canondata/test.test_purebench_smoke_blocks_/out
index f02ba5273b7..8032834491e 100644
--- a/yql/essentials/tools/purebench/test/canondata/test.test_purebench_smoke_blocks_/out
+++ b/yql/essentials/tools/purebench/test/canondata/test.test_purebench_smoke_blocks_/out
@@ -1,2 +1,2 @@
-Elapsed: <DURATION>s
+Benchmark completed: <ITERATIONS> iterations for <DURATION>s
 Bench score: <SCORE>
diff --git a/yql/essentials/tools/purebench/test/canondata/test.test_purebench_smoke_scalar_/err b/yql/essentials/tools/purebench/test/canondata/test.test_purebench_smoke_scalar_/err
index 6b3cdf452ad..0425494f361 100644
--- a/yql/essentials/tools/purebench/test/canondata/test.test_purebench_smoke_scalar_/err
+++ b/yql/essentials/tools/purebench/test/canondata/test.test_purebench_smoke_scalar_/err
@@ -21,3 +21,5 @@ Generated data size: 12000000
 }
 Dry run of test sql...
 Run benchmark...
+Score calibration...
+Calibration completed: <ITERATIONS> iterations for <DURATION>s
diff --git a/yql/essentials/tools/purebench/test/canondata/test.test_purebench_smoke_scalar_/out b/yql/essentials/tools/purebench/test/canondata/test.test_purebench_smoke_scalar_/out
index f02ba5273b7..8032834491e 100644
--- a/yql/essentials/tools/purebench/test/canondata/test.test_purebench_smoke_scalar_/out
+++ b/yql/essentials/tools/purebench/test/canondata/test.test_purebench_smoke_scalar_/out
@@ -1,2 +1,2 @@
-Elapsed: <DURATION>s
+Benchmark completed: <ITERATIONS> iterations for <DURATION>s
 Bench score: <SCORE>
diff --git a/yql/essentials/tools/purebench/test/test.py b/yql/essentials/tools/purebench/test/test.py
index 2ca966f4135..aa96bc7c578 100644
--- a/yql/essentials/tools/purebench/test/test.py
+++ b/yql/essentials/tools/purebench/test/test.py
@@ -7,19 +7,26 @@ PUREBENCH = yatest.common.build_path('yql/essentials/tools/purebench/purebench')
 
 def run_purebench_test(cmdline):
     result = yatest.common.execute(cmdline, text=True, check_exit_code=True)
-    # Mask elapsed time and duration, since both can change in
+    # Mask benchmark score and duration, since both can change in
     # different environments.
     stdout = result.stdout
-    stdout = re.sub(r'(Elapsed: )(\d+([.]\d+)?)', r'\1<DURATION>', stdout)
-    stdout = re.sub(r'(Bench score: )(\d+([.]\d+)?|nan)', r'\1<SCORE>', stdout)
+    stdout = re.sub(
+        r'(Benchmark completed: )(\d+)( iterations for )(\d+([.]\d+)?)', r'\1<ITERATIONS>\3<DURATION>', stdout
+    )
+    stdout = re.sub(r'(Bench score: )(\d+([.]\d+(e+\d+)?)?|nan)', r'\1<SCORE>', stdout)
     # Dump the masked stdout.
     outfile = yatest.common.output_path('out')
     with open(outfile, "w") as dump:
         dump.write(stdout)
-    # Dump the raw stderr, since all the diagnostic is precise.
+    # Mask calibration stats, since it can change either.
+    stderr = result.stderr
+    stderr = re.sub(
+        r'(Calibration completed: )(\d+)( iterations for )(\d+([.]\d+)?)', r'\1<ITERATIONS>\3<DURATION>', stderr
+    )
+    # Dump the masked stderr.
     errfile = yatest.common.output_path('err')
     with open(errfile, "w") as dump:
-        dump.write(result.stderr)
+        dump.write(stderr)
     # Canonize both stdout and stderr dumps locally.
     return [
         yatest.common.canonical_file(outfile, local=True),
@@ -41,6 +48,12 @@ def test_purebench_smoke(useBlocks):
             '--ndebug',
             '--repeats',
             '1',
+            '--calibrate',
+            '1',
+            '--repeat-time',
+            '0',
+            '--calibrate-time',
+            '0',
             '--blocks-engine',
             'force' if useBlocks else 'disable',
         ]
@@ -56,7 +69,20 @@ def test_purebench_smoke(useBlocks):
     ],
 )
 def test_purebench_langver(langVer):
-    cmdline = [PUREBENCH, '--ndebug', '--repeats', '0', '-t', 'SELECT CurrentLanguageVersion()']
+    cmdline = [
+        PUREBENCH,
+        '--ndebug',
+        '--repeats',
+        '0',
+        '--repeat-time',
+        '0',
+        '--calibrate',
+        '0',
+        '--calibrate-time',
+        '0',
+        '-t',
+        'SELECT CurrentLanguageVersion()',
+    ]
     if langVer:
         cmdline.extend(['--langver', langVer])
     return run_purebench_test(cmdline)
@@ -78,6 +104,12 @@ def test_purebench_inc(useBlocks):
             '--print-expr',
             '--repeats',
             '1',
+            '--calibrate',
+            '1',
+            '--repeat-time',
+            '0',
+            '--calibrate-time',
+            '0',
             '-w',
             '0',
             '--blocks-engine',
@@ -104,6 +136,12 @@ def test_purebench_inc_for_42(useBlocks):
             '--print-expr',
             '--repeats',
             '1',
+            '--calibrate',
+            '1',
+            '--repeat-time',
+            '0',
+            '--calibrate-time',
+            '0',
             '-w',
             '0',
             '--blocks-engine',
diff --git a/yql/essentials/tools/purebench/test/ya.make b/yql/essentials/tools/purebench/test/ya.make
index 20e1cdaf29e..0555a44bbf4 100644
--- a/yql/essentials/tools/purebench/test/ya.make
+++ b/yql/essentials/tools/purebench/test/ya.make
@@ -1,7 +1,7 @@
 PY3TEST()
 
 SIZE(MEDIUM)
-TIMEOUT(180)
+TIMEOUT(240)
 
 TEST_SRCS(
     test.py