Revert D95149629: feat(expr-eval):Adaptive per-function CPU sampling for Velox expression evaluation

Summary:
Reverting D95149629 because it introduced a persistently failing test
`ExprStatsTest.adaptiveCpuSamplingPerFunctionRates` that has been breaking
OSS CI (Linux Build using GCC) on every merge-to-main commit since landing.

This caused 6+ consecutive red builds on the Velox main branch starting
from commit 793f13f (Mar 31 18:28 UTC).

The feature can be re-landed after the test is fixed.

Differential Revision: D99117623

Author: kgpai

velox/core/QueryConfig.h

@@ -87,25 +87,6 @@ class QueryConfig {
   static constexpr const char* kExprTrackCpuUsageForFunctions =
       "expression.track_cpu_usage_for_functions";
 
-  /// Enables adaptive per-function CPU usage sampling. When enabled, each
-  /// function is calibrated over the first 6 batches (1 warmup + 5
-  /// calibration) to measure the overhead of CPU tracking (clock_gettime).
-  /// Functions where tracking overhead exceeds
-  /// kExprAdaptiveCpuSamplingMaxOverheadPct are automatically sampled at a
-  /// rate proportional to their overhead. Functions with low overhead are
-  /// always tracked. Disabled by default.
-  static constexpr const char* kExprAdaptiveCpuSampling =
-      "expression.adaptive_cpu_sampling";
-
-  /// Maximum acceptable overhead percentage for CPU tracking per function.
-  /// Used with kExprAdaptiveCpuSampling. Functions whose CPU tracking overhead
-  /// exceeds this threshold are sampled at a rate of
-  /// ceil(overhead_pct / max_overhead_pct). For example, with max_overhead=1.0,
-  /// a function with 70% tracking overhead is sampled every 70th batch.
-  /// Default: 1.0 (1% overhead target).
-  static constexpr const char* kExprAdaptiveCpuSamplingMaxOverheadPct =
-      "expression.adaptive_cpu_sampling_max_overhead_pct";
-
   /// Controls whether non-deterministic expressions are deduplicated during
   /// compilation. This is intended for testing and debugging purposes. By
   /// default, this is set to true to preserve standard behavior. If set to
@@ -1406,14 +1387,6 @@ class QueryConfig {
     return get<std::string>(kExprTrackCpuUsageForFunctions, "");
   }
 
-  bool exprAdaptiveCpuSampling() const {
-    return get<bool>(kExprAdaptiveCpuSampling, false);
-  }
-
-  double exprAdaptiveCpuSamplingMaxOverheadPct() const {
-    return get<double>(kExprAdaptiveCpuSamplingMaxOverheadPct, 1.0);
-  }
-
   bool exprDedupNonDeterministic() const {
     return get<bool>(kExprDedupNonDeterministic, true);
   }

velox/docs/configs.rst

@@ -286,22 +286,6 @@ Expression Evaluation Configuration
        ``expression.track_cpu_usage`` is set to false. Function names are case-insensitive and will be normalized
        to lowercase. This allows fine-grained control over CPU tracking overhead when only specific functions need to
        be monitored.
-   * - expression.adaptive_cpu_sampling
-     - boolean
-     - false
-     - Enables adaptive per-function CPU usage sampling. Each function is calibrated over 6 batches (1 warmup + 5
-       calibration) to measure the overhead of CPU tracking (clock_gettime) relative to the function's execution time.
-       The timer overhead is measured once per ExprSet and shared across all functions. Functions where tracking overhead
-       is acceptable are always tracked; functions where overhead exceeds ``expression.adaptive_cpu_sampling_max_overhead_pct``
-       are sampled at a rate proportional to their overhead. Sampled timing stats are extrapolated to approximate
-       full-population values.
-   * - expression.adaptive_cpu_sampling_max_overhead_pct
-     - float
-     - 1.0
-     - Maximum acceptable CPU tracking overhead percentage per function, used with ``expression.adaptive_cpu_sampling``.
-       Functions whose tracking overhead exceeds this threshold are sampled at a rate of
-       ceil(overhead_pct / max_overhead_pct). For example, with max_overhead=1.0, a function with 70% tracking overhead
-       is sampled every 70th batch, bounding its effective overhead to ~1%. Must be greater than 0.
    * - legacy_cast
      - bool
      - false

velox/expression/EvalCtx.h

@@ -539,34 +539,6 @@ class EvalCtx {
     return execCtx_->optimizationParams().dictionaryMemoizationEnabled;
   }
 
-  /// Returns true if adaptive per-function CPU sampling is enabled.
-  bool adaptiveCpuSamplingEnabled() const {
-    return adaptiveCpuSamplingEnabled_;
-  }
-
-  void setAdaptiveCpuSamplingEnabled(bool enabled) {
-    adaptiveCpuSamplingEnabled_ = enabled;
-  }
-
-  /// Returns the maximum acceptable overhead pct for adaptive sampling.
-  double adaptiveCpuSamplingMaxOverheadPct() const {
-    return adaptiveCpuSamplingMaxOverheadPct_;
-  }
-
-  void setAdaptiveCpuSamplingMaxOverheadPct(double pct) {
-    adaptiveCpuSamplingMaxOverheadPct_ = pct;
-  }
-
-  /// Returns the measured CpuWallTimer overhead in nanoseconds (per
-  /// invocation). Measured once per ExprSet and shared across all Expr nodes.
-  uint64_t timerOverheadNanos() const {
-    return timerOverheadNanos_;
-  }
-
-  void setTimerOverheadNanos(uint64_t nanos) {
-    timerOverheadNanos_ = nanos;
-  }
-
   /// Returns the maximum number of distinct inputs to cache results for in a
   /// given shared subexpression.
   uint32_t maxSharedSubexprResultsCached() const {
@@ -638,15 +610,6 @@ class EvalCtx {
   // If 'captureErrorDetails()' is false, stores flags indicating which rows had
   // errors without storing actual exceptions.
   EvalErrorsPtr errors_;
-
-  // Whether adaptive per-function CPU sampling is enabled.
-  bool adaptiveCpuSamplingEnabled_{false};
-
-  // Maximum acceptable overhead percentage for adaptive CPU sampling.
-  double adaptiveCpuSamplingMaxOverheadPct_{1.0};
-
-  // Measured CpuWallTimer overhead (nanos per invocation), shared across Exprs.
-  uint64_t timerOverheadNanos_{0};
 };
 
 /// Utility wrapper struct that is used to temporarily reset the value of the

velox/expression/Expr.cpp

@@ -16,7 +16,6 @@
 #include <boost/lexical_cast.hpp>
 #include <boost/uuid/uuid_generators.hpp>
 #include <boost/uuid/uuid_io.hpp>
-#include <cmath>
 
 #include "velox/common/base/Exceptions.h"
 #include "velox/common/base/Fs.h"
@@ -1588,94 +1587,13 @@ bool Expr::applyFunctionWithPeeling(
   return true;
 }
 
-std::unique_ptr<CpuWallTimer> Expr::cpuWallTimer(const EvalCtx& context) {
-  // 1. Compile-time tracking (set via trackCpuUsage_) always wins.
-  if (trackCpuUsage_) {
-    return std::make_unique<CpuWallTimer>(stats_.timing);
-  }
-
-  // 2. Adaptive per-function sampling.
-  if (context.adaptiveCpuSamplingEnabled()) {
-    switch (adaptiveState_) {
-      case AdaptiveCpuSamplingState::kWarmup:
-        // Warmup batch: just run the function, no timing.
-        return nullptr;
-      case AdaptiveCpuSamplingState::kCalibrating: {
-        // Measure function execution time (without CpuWallTimer).
-        // Timer overhead is measured once per ExprSet and shared via EvalCtx.
-        calibrationStopWatch_.emplace();
-        return nullptr;
-      }
-      case AdaptiveCpuSamplingState::kAlwaysTrack:
-        return std::make_unique<CpuWallTimer>(stats_.timing);
-      case AdaptiveCpuSamplingState::kSampling:
-        if (++adaptiveSamplingCounter_ % adaptiveSamplingRate_ == 0) {
-          return std::make_unique<CpuWallTimer>(stats_.timing);
-        }
-        return nullptr;
-    }
-  }
-
-  return nullptr;
-}
-
-void Expr::finalizeAdaptiveCalibration(
-    double maxOverheadPct,
-    uint64_t timerOverheadNanos) {
-  switch (adaptiveState_) {
-    case AdaptiveCpuSamplingState::kWarmup: {
-      adaptiveState_ = AdaptiveCpuSamplingState::kCalibrating;
-      break;
-    }
-    case AdaptiveCpuSamplingState::kCalibrating: {
-      calibrationFunctionWallNanos_ +=
-          calibrationStopWatch_->elapsed().wallNanos;
-      calibrationStopWatch_.reset();
-
-      if (++calibrationBatchCount_ < kCalibrationBatches) {
-        break;
-      }
-
-      // Use the shared timer overhead measurement, scaled by calibration
-      // batch count. The overhead per invocation is a platform constant
-      // measured once per ExprSet.
-      auto totalTimerOverhead = timerOverheadNanos * calibrationBatchCount_;
-
-      if (calibrationFunctionWallNanos_ > 0 && maxOverheadPct > 0) {
-        double overheadPct = 100.0 * static_cast<double>(totalTimerOverhead) /
-            static_cast<double>(calibrationFunctionWallNanos_);
-
-        if (overheadPct > maxOverheadPct) {
-          adaptiveSamplingRate_ =
-              static_cast<uint32_t>(std::ceil(overheadPct / maxOverheadPct));
-          // Start counter at rate-1 so the first post-calibration batch is
-          // always timed (++counter hits rate, which passes % rate == 0).
-          adaptiveSamplingCounter_ = adaptiveSamplingRate_ - 1;
-          adaptiveState_ = AdaptiveCpuSamplingState::kSampling;
-        } else {
-          adaptiveState_ = AdaptiveCpuSamplingState::kAlwaysTrack;
-        }
-      } else {
-        // Function ~0ns — timer dominates. Aggressive sampling.
-        adaptiveSamplingRate_ = 100;
-        adaptiveSamplingCounter_ = adaptiveSamplingRate_ - 1;
-        adaptiveState_ = AdaptiveCpuSamplingState::kSampling;
-      }
-      break;
-    }
-    default:
-      VELOX_UNREACHABLE(
-          "Unexpected adaptive sampling state in finalizeAdaptiveCalibration");
-  }
-}
-
 void Expr::applyFunction(
     const SelectivityVector& rows,
     EvalCtx& context,
     VectorPtr& result) {
   stats_.numProcessedVectors += 1;
   stats_.numProcessedRows += rows.countSelected();
-  auto timer = cpuWallTimer(context);
+  auto timer = cpuWallTimer();
 
   computeIsAsciiForInputs(vectorFunction_.get(), inputValues_, rows);
   auto isAscii = type()->isVarchar()
@@ -1715,14 +1633,6 @@ void Expr::applyFunction(
     result->asUnchecked<SimpleVector<StringView>>()->setIsAscii(
         isAscii.value(), rows);
   }
-
-  // Only do Adaptive Calibration if the adaptive sampling is on and we are in
-  // warmup or calibrating state.
-  if (context.adaptiveCpuSamplingEnabled() && isCalibrating()) {
-    finalizeAdaptiveCalibration(
-        context.adaptiveCpuSamplingMaxOverheadPct(),
-        context.timerOverheadNanos());
-  }
 }
 
 void Expr::evalSpecialFormWithStats(
@@ -1731,17 +1641,9 @@ void Expr::evalSpecialFormWithStats(
     VectorPtr& result) {
   stats_.numProcessedVectors += 1;
   stats_.numProcessedRows += rows.countSelected();
-  auto timer = cpuWallTimer(context);
+  auto timer = cpuWallTimer();
 
   evalSpecialForm(rows, context, result);
-
-  // Only do Adaptive Calibration if the adaptive sampling is on and we are in
-  // warmup or calibrating state.
-  if (context.adaptiveCpuSamplingEnabled() && isCalibrating()) {
-    finalizeAdaptiveCalibration(
-        context.adaptiveCpuSamplingMaxOverheadPct(),
-        context.timerOverheadNanos());
-  }
 }
 
 namespace {
@@ -1971,14 +1873,7 @@ ExprSet::ExprSet(
     core::ExecCtx* execCtx,
     bool enableConstantFolding,
     bool lazyDereference)
-    : execCtx_(execCtx),
-      lazyDereference_(lazyDereference),
-      adaptiveCpuSampling_(
-          execCtx->queryCtx()->queryConfig().exprAdaptiveCpuSampling()),
-      adaptiveCpuSamplingMaxOverheadPct_(
-          execCtx->queryCtx()
-              ->queryConfig()
-              .exprAdaptiveCpuSamplingMaxOverheadPct()) {
+    : execCtx_(execCtx), lazyDereference_(lazyDereference) {
   exprs_ = compileExpressions(sources, execCtx, this, enableConstantFolding);
   if (lazyDereference_) {
     validateLazyDereference(exprs_);
@@ -1991,24 +1886,6 @@ ExprSet::ExprSet(
 }
 
 namespace {
-
-/// If the expression is in adaptive sampling mode, extrapolate timing stats
-/// to approximate full-population values. Otherwise, return raw stats.
-exec::ExprStats adjustStats(const exec::Expr& expr) {
-  if (expr.isAdaptiveSampling() && expr.stats().timing.count > 0) {
-    exec::ExprStats adjusted = expr.stats();
-    double ratio = static_cast<double>(adjusted.numProcessedVectors) /
-        static_cast<double>(adjusted.timing.count);
-    adjusted.timing.cpuNanos = static_cast<uint64_t>(
-        static_cast<double>(adjusted.timing.cpuNanos) * ratio);
-    adjusted.timing.wallNanos = static_cast<uint64_t>(
-        static_cast<double>(adjusted.timing.wallNanos) * ratio);
-    adjusted.timing.count = adjusted.numProcessedVectors;
-    return adjusted;
-  }
-  return expr.stats();
-}
-
 void addStats(
     const exec::Expr& expr,
     std::unordered_map<std::string, exec::ExprStats>& stats,
@@ -2027,7 +1904,7 @@ void addStats(
   bool emptyStats =
       !expr.stats().numProcessedRows && !expr.stats().defaultNullRowsSkipped;
   if (!emptyStats && !excludeSplFormExpr) {
-    stats[expr.name()].add(adjustStats(expr));
+    stats[expr.name()].add(expr.stats());
   }
 
   for (const auto& input : expr.inputs()) {
@@ -2139,24 +2016,6 @@ void printInputAndExprs(
 }
 } // namespace
 
-void ExprSet::initializeAdaptiveCpuSampling(EvalCtx& context) {
-  context.setAdaptiveCpuSamplingEnabled(true);
-  context.setAdaptiveCpuSamplingMaxOverheadPct(
-      adaptiveCpuSamplingMaxOverheadPct_);
-
-  // Measure CpuWallTimer overhead once per ExprSet (platform constant).
-  if (!timerOverheadMeasured_) {
-    CpuWallTiming dummyTiming;
-    DeltaCpuWallTimeStopWatch overheadWatch;
-    {
-      auto dummy = std::make_unique<CpuWallTimer>(dummyTiming);
-    }
-    timerOverheadNanos_ = overheadWatch.elapsed().wallNanos;
-    timerOverheadMeasured_ = true;
-  }
-  context.setTimerOverheadNanos(timerOverheadNanos_);
-}
-
 void ExprSet::eval(
     int32_t begin,
     int32_t end,
@@ -2170,11 +2029,6 @@ void ExprSet::eval(
     clearSharedSubexprs();
   }
 
-  // Apply adaptive per-function CPU sampling if configured.
-  if (adaptiveCpuSampling_) {
-    initializeAdaptiveCpuSampling(context);
-  }
-
   if (!lazyDereference_) {
     // Make sure LazyVectors, referenced by multiple expressions, are loaded for
     // all the "rows".