linkrs_query/optimizer/stats/feedback/

selectivity.rs

//! Selective Correction Module
//!
//! Provide the functionality to dynamically adjust selective estimates based on historical performance feedback.
//! Uses an exponentially weighted moving average (EWMA) algorithm to correct selectivity estimates.

use parking_lot::RwLock;
use std::collections::HashMap;

/// Feedback-driven selective correction
///
/// Dynamically adjust the selective estimates based on historical feedback from executions.
/// Uses an exponentially weighted moving average (EWMA) algorithm.
///
/// # Example
/// ```
/// use linkrs_query::optimizer::stats::feedback::selectivity::FeedbackDrivenSelectivity;
///
/// let mut feedback = FeedbackDrivenSelectivity::new(0.1);
/// assert_eq!(feedback.estimated_selectivity(), 0.1);
///
/// // Update the feedback
/// feedback.update_with_feedback(0.15);
/// feedback.update_with_feedback(0.12);
///
/// // Obtain the corrected selective information
/// let corrected = feedback.corrected_selectivity();
/// assert!(corrected > 0.0 && corrected <= 0.99);
/// ```
#[derive(Debug, Clone)]
pub struct FeedbackDrivenSelectivity {
    /// Original estimate of selectivity
    estimated_selectivity: f64,
    /// Historical Selectivity (Sliding Window Average)
    actual_selectivity_ewma: f64,
    /// Correction factor
    correction_factor: f64,
    /// Number of feedbacks
    feedback_count: u64,
    /// EWMA smoothing factor
    alpha: f64,
    /// Minimum correction factor
    min_correction: f64,
    /// Maximum correction factor
    max_correction: f64,
    /// Selective upper limit (default 0.99)
    selectivity_cap: f64,
    /// Cumulative estimated error (used for calculating error statistics)
    cumulative_estimation_error: f64,
    /// Sum of the squared errors (used to calculate the standard deviation)
    error_sum_squares: f64,
    /// Total measured input rows across applied samples. Small samples
    /// move the factor conservatively (see [`Self::update_with_measured_ratio`]).
    total_input_rows: u64,
    /// One staged outlier sample awaiting corroboration by an independent
    /// execution before it may move the factor.
    pending_outlier: Option<MeasuredSample>,
}

/// One measured execution sample: observed local-parameter ratio plus the
/// input rows it was measured over.
#[derive(Debug, Clone, Copy)]
struct MeasuredSample {
    ratio: f64,
    input_rows: u64,
}

/// Samples deviating from the learned factor by more than this are staged
/// instead of applied: a lone distorted measurement (truncated run,
/// misattributed upstream error) must not yank the correction.
const OUTLIER_DEVIATION: f64 = 4.0;
/// A staged sample is adopted once an independent execution agrees with it
/// within this factor.
const CORROBORATION_AGREEMENT: f64 = 2.0;
/// Input rows at which a sample earns the full EWMA step; smaller samples
/// move the factor proportionally less. Keeps tiny measurements from
/// flipping corrections while letting table-scale runs learn normally.
const FULL_WEIGHT_INPUT_ROWS: f64 = 1000.0;

impl FeedbackDrivenSelectivity {
    /// Create a new feedback-driven, selective estimation method
    pub fn new(estimated_selectivity: f64) -> Self {
        Self {
            estimated_selectivity,
            actual_selectivity_ewma: estimated_selectivity,
            correction_factor: 1.0,
            feedback_count: 0,
            alpha: 0.3,
            min_correction: 0.1,
            max_correction: 10.0,
            selectivity_cap: 0.99,
            cumulative_estimation_error: 0.0,
            error_sum_squares: 0.0,
            total_input_rows: 0,
            pending_outlier: None,
        }
    }

    /// Create using custom parameters
    pub fn with_params(
        estimated_selectivity: f64,
        alpha: f64,
        min_correction: f64,
        max_correction: f64,
    ) -> Self {
        Self {
            estimated_selectivity,
            actual_selectivity_ewma: estimated_selectivity,
            correction_factor: 1.0,
            feedback_count: 0,
            alpha,
            min_correction,
            max_correction,
            selectivity_cap: 0.99,
            cumulative_estimation_error: 0.0,
            error_sum_squares: 0.0,
            total_input_rows: 0,
            pending_outlier: None,
        }
    }

    /// Obtain the original estimate for selectivity.
    pub fn estimated_selectivity(&self) -> f64 {
        self.estimated_selectivity
    }

    /// Track a fresh uncorrected estimate while preserving the learned
    /// correction factor and feedback history.
    pub fn set_estimated_selectivity(&mut self, estimated: f64) {
        if estimated > 0.0 && estimated.is_finite() {
            self.estimated_selectivity = estimated;
            self.actual_selectivity_ewma = estimated * self.correction_factor;
        }
    }

    /// Obtain the corrected version of the selective content.
    pub fn corrected_selectivity(&self) -> f64 {
        self.corrected_for_estimate(self.estimated_selectivity)
    }

    /// Correct a fresh estimate with the learned factor without touching the
    /// stored baseline, so snapshot reads stay free of planning-order drift.
    pub fn corrected_for_estimate(&self, estimated: f64) -> f64 {
        (estimated * self.correction_factor)
            .clamp(self.min_correction * estimated, self.selectivity_cap)
    }

    /// Obtain the correction factor
    pub fn correction_factor(&self) -> f64 {
        self.correction_factor
    }

    /// Number of times feedback was obtained
    pub fn feedback_count(&self) -> u64 {
        self.feedback_count
    }

    /// Total measured input rows across applied samples.
    pub fn total_input_rows(&self) -> u64 {
        self.total_input_rows
    }

    /// Obtaining the confidence level for the selected estimate
    ///
    /// The confidence level is calculated based on the number of feedbacks received; the more feedback, the higher the confidence level.
    /// Return value range: 0.0 - 1.0
    pub fn estimation_confidence(&self) -> f64 {
        // Calculate the confidence using the sigmoid function.
        // The confidence level is close to 0.9 when the number of feedbacks reaches 100
        let x = self.feedback_count as f64 * 0.1;
        1.0 / (1.0 + (-x).exp())
    }

    /// Obtain the average estimated error.
    pub fn avg_estimation_error(&self) -> f64 {
        if self.feedback_count == 0 {
            return 1.0;
        }
        self.cumulative_estimation_error / self.feedback_count as f64
    }

    /// Obtain the standard deviation of the estimated error
    pub fn error_std_dev(&self) -> f64 {
        if self.feedback_count < 2 {
            return 0.0;
        }
        let n = self.feedback_count as f64;
        let mean = self.cumulative_estimation_error / n;
        let variance = (self.error_sum_squares / n) - (mean * mean);
        variance.max(0.0).sqrt()
    }

    /// Update the correction factor (based on the new feedback).
    ///
    /// Uses an exponentially weighted moving average (EWMA) algorithm.
    pub fn update_with_feedback(&mut self, actual_selectivity: f64) {
        if self.estimated_selectivity <= 0.0 {
            return;
        }

        let ratio = actual_selectivity / self.estimated_selectivity;

        // Calculate the current estimated error.
        let estimated = self.corrected_selectivity();
        let error = (actual_selectivity - estimated).abs();
        self.cumulative_estimation_error += error;
        self.error_sum_squares += error * error;

        // Use the EWMA to update the correction factor.
        self.correction_factor = (1.0 - self.alpha) * self.correction_factor + self.alpha * ratio;

        // Limit the range of the correction factors to avoid excessive correction.
        self.correction_factor = self
            .correction_factor
            .clamp(self.min_correction, self.max_correction);

        // Updated version of the practical selective EWMA (Exponential Moving Average)
        self.actual_selectivity_ewma =
            (1.0 - self.alpha) * self.actual_selectivity_ewma + self.alpha * actual_selectivity;

        self.feedback_count += 1;
    }

    /// Update the correction factor from an observed estimated-vs-actual row
    /// ratio (actual_rows / estimated_rows).
    ///
    /// Unlike [`update_with_feedback`], which receives an absolute
    /// selectivity, this receives the correction ratio directly so the
    /// feedback loop does not need to know the operator's input cardinality.
    pub fn update_with_ratio(&mut self, ratio: f64) {
        if self.estimated_selectivity <= 0.0 || !ratio.is_finite() {
            return;
        }

        let error = (ratio - self.correction_factor).abs();
        self.cumulative_estimation_error += error;
        self.error_sum_squares += error * error;

        // Use the EWMA to update the correction factor.
        self.correction_factor = (1.0 - self.alpha) * self.correction_factor + self.alpha * ratio;

        // Limit the range of the correction factors to avoid excessive correction.
        self.correction_factor = self
            .correction_factor
            .clamp(self.min_correction, self.max_correction);

        // Keep the smoothed absolute selectivity consistent with the factor.
        self.actual_selectivity_ewma = self.estimated_selectivity * self.correction_factor;

        self.feedback_count += 1;
    }

    /// Update the correction factor from a measured local-parameter ratio
    /// over a known number of input rows.
    ///
    /// This is the feedback loop's write path. Unlike [`update_with_ratio`],
    /// the ratio must already isolate the operator-local parameter
    /// (e.g. observed-vs-estimated *selectivity*, not output rows), because
    /// output ratios conflate upstream cardinality error. Two guards keep
    /// distorted measurements out of the factor:
    ///
    /// - samples deviating from the learned factor by more than
    ///   [`OUTLIER_DEVIATION`] are staged until an independent execution
    ///   corroborates them (the first sample has no baseline and applies);
    /// - the EWMA step scales with measured input rows, so tiny samples
    ///   move the factor conservatively.
    pub fn update_with_measured_ratio(&mut self, ratio: f64, input_rows: u64) {
        if self.estimated_selectivity <= 0.0 || !ratio.is_finite() || ratio < 0.0 {
            return;
        }
        let sample = MeasuredSample { ratio, input_rows };
        if self.feedback_count > 0 && self.deviates(sample.ratio) {
            if let Some(staged) = self.pending_outlier {
                if agrees(staged.ratio, sample.ratio) {
                    self.pending_outlier = None;
                    self.apply_measured(staged);
                    self.apply_measured(sample);
                } else {
                    self.pending_outlier = Some(sample);
                }
            } else {
                self.pending_outlier = Some(sample);
            }
            return;
        }
        self.apply_measured(sample);
    }

    /// Relative deviation of a sample from the learned factor.
    fn deviates(&self, ratio: f64) -> bool {
        let factor = self.correction_factor;
        if factor <= 0.0 || ratio <= 0.0 {
            return ratio != factor;
        }
        ratio.max(factor) / ratio.min(factor) > OUTLIER_DEVIATION
    }

    /// Fold one corroborated sample into the factor with a step
    /// proportional to its measured input rows.
    fn apply_measured(&mut self, sample: MeasuredSample) {
        let weight = (sample.input_rows as f64).max(1.0);
        let step = self.alpha * weight / (weight + FULL_WEIGHT_INPUT_ROWS);
        let error = (sample.ratio - self.correction_factor).abs();
        self.cumulative_estimation_error += error;
        self.error_sum_squares += error * error;
        self.correction_factor = (1.0 - step) * self.correction_factor + step * sample.ratio;
        self.correction_factor = self
            .correction_factor
            .clamp(self.min_correction, self.max_correction);
        self.actual_selectivity_ewma = self.estimated_selectivity * self.correction_factor;
        self.total_input_rows += sample.input_rows;
        self.feedback_count += 1;
    }

    /// Batch update of feedback
    pub fn update_with_batch(&mut self, actual_selectivities: &[f64]) {
        for &selectivity in actual_selectivities {
            self.update_with_feedback(selectivity);
        }
    }

    /// Reset the correction factor
    pub fn reset_correction(&mut self) {
        self.correction_factor = 1.0;
        self.actual_selectivity_ewma = self.estimated_selectivity;
        self.feedback_count = 0;
        self.cumulative_estimation_error = 0.0;
        self.error_sum_squares = 0.0;
        self.total_input_rows = 0;
        self.pending_outlier = None;
    }

    /// Setting the EWMA smoothing factor
    pub fn set_alpha(&mut self, alpha: f64) {
        self.alpha = alpha.clamp(0.0, 1.0);
    }

    /// Set the correction range
    pub fn set_correction_range(&mut self, min: f64, max: f64) {
        self.min_correction = min.max(0.01);
        self.max_correction = max.max(self.min_correction);
    }

    /// Setting a selective upper limit
    pub fn set_selectivity_cap(&mut self, cap: f64) {
        self.selectivity_cap = cap.clamp(0.5, 1.0);
    }
}

impl Default for FeedbackDrivenSelectivity {
    fn default() -> Self {
        Self::new(0.1)
    }
}

/// Selective Feedback Manager
///
/// Managing selective feedback based on multiple conditions.
///
/// # Examples
/// ```
/// use linkrs_query::optimizer::stats::feedback::selectivity::SelectivityFeedbackManager;
///
/// let manager = SelectivityFeedbackManager::new();
/// manager.register_condition("age > 25".to_string(), 0.3);
///
/// let corrected = manager.get_corrected_selectivity("age > 25");
/// assert!(corrected.is_some());
/// ```
#[derive(Debug)]
pub struct SelectivityFeedbackManager {
    /// Mapping from condition keys to selective correction
    feedbacks: RwLock<HashMap<String, FeedbackDrivenSelectivity>>,
    /// Default EWMA smoothing factor
    default_alpha: f64,
    /// Default minimum correction factor
    default_min_correction: f64,
    /// Default maximum correction factor
    default_max_correction: f64,
}

impl SelectivityFeedbackManager {
    /// Create a new feedback manager.
    pub fn new() -> Self {
        Self {
            feedbacks: RwLock::new(HashMap::new()),
            default_alpha: 0.3,
            default_min_correction: 0.1,
            default_max_correction: 10.0,
        }
    }

    /// Created using custom parameters
    pub fn with_params(alpha: f64, min_correction: f64, max_correction: f64) -> Self {
        Self {
            feedbacks: RwLock::new(HashMap::new()),
            default_alpha: alpha,
            default_min_correction: min_correction,
            default_max_correction: max_correction,
        }
    }

    /// Selective estimation of registration requirements
    ///
    /// Registers the condition only when the key is not present yet, so
    /// repeated estimation of the same predicate does not reset previously
    /// learned corrections.
    pub fn register_condition(&self, key: String, estimated_selectivity: f64) {
        let feedback = FeedbackDrivenSelectivity::with_params(
            estimated_selectivity,
            self.default_alpha,
            self.default_min_correction,
            self.default_max_correction,
        );
        self.feedbacks.write().entry(key).or_insert(feedback);
    }

    /// Update the baseline estimate of a registered condition while
    /// preserving its learned correction factor, so baselines track fresh
    /// statistics instead of freezing at first sight. Returns false when
    /// the key is unknown.
    pub fn refresh_estimated(&self, key: &str, estimated_selectivity: f64) -> bool {
        let mut feedbacks = self.feedbacks.write();
        if let Some(feedback) = feedbacks.get_mut(key) {
            feedback.set_estimated_selectivity(estimated_selectivity);
            true
        } else {
            false
        }
    }

    /// Getting corrected selectivity
    pub fn get_corrected_selectivity(&self, key: &str) -> Option<f64> {
        self.feedbacks
            .read()
            .get(key)
            .map(|f| f.corrected_selectivity())
    }

    /// Capture an immutable snapshot of the current learned corrections for
    /// one planning decision to read through without mutating shared state.
    pub fn snapshot(&self) -> SelectivitySnapshot {
        SelectivitySnapshot {
            corrections: self.feedbacks.read().clone(),
        }
    }

    /// Update feedback
    pub fn update_feedback(&self, key: &str, actual_selectivity: f64) -> bool {
        let mut feedbacks = self.feedbacks.write();
        if let Some(feedback) = feedbacks.get_mut(key) {
            feedback.update_with_feedback(actual_selectivity);
            true
        } else {
            false
        }
    }

    /// Update feedback from an estimated-vs-actual row ratio.
    ///
    /// The ratio is the correction factor `actual_rows / estimated_rows`;
    /// returns `false` when the key has not been registered.
    pub fn update_feedback_ratio(&self, key: &str, ratio: f64) -> bool {
        let mut feedbacks = self.feedbacks.write();
        if let Some(feedback) = feedbacks.get_mut(key) {
            feedback.update_with_ratio(ratio);
            true
        } else {
            false
        }
    }

    /// Update feedback from a measured local-parameter ratio over a known
    /// number of input rows (the feedback loop's write path).
    ///
    /// Returns `false` when the key has not been registered; staged
    /// outliers still return `true` (the sample was recorded, just not
    /// applied yet).
    pub fn update_measured_ratio(&self, key: &str, ratio: f64, input_rows: u64) -> bool {
        let mut feedbacks = self.feedbacks.write();
        if let Some(feedback) = feedbacks.get_mut(key) {
            feedback.update_with_measured_ratio(ratio, input_rows);
            true
        } else {
            false
        }
    }

    /// Remove all feedback whose key starts with `space_prefix` (e.g.
    /// `"myspace:"`), returning the number of removed entries.
    pub fn remove_feedback_by_space(&self, space_prefix: &str) -> usize {
        let mut feedbacks = self.feedbacks.write();
        let mut removed = 0;
        feedbacks.retain(|key, _| {
            let keep = !key.starts_with(space_prefix);
            if !keep {
                removed += 1;
            }
            keep
        });
        removed
    }

    /// Batch Update Feedback
    pub fn update_feedback_batch(&self, updates: &[(String, f64)]) {
        let mut feedbacks = self.feedbacks.write();
        for (key, actual_selectivity) in updates {
            if let Some(feedback) = feedbacks.get_mut(key) {
                feedback.update_with_feedback(*actual_selectivity);
            }
        }
    }

    /// Obtain feedback information
    pub fn get_feedback(&self, key: &str) -> Option<FeedbackDrivenSelectivity> {
        self.feedbacks.read().get(key).cloned()
    }

    /// Obtain all feedback keys.
    pub fn get_all_keys(&self) -> Vec<String> {
        self.feedbacks.read().keys().cloned().collect()
    }

    /// Clear all feedback.
    pub fn clear_all(&self) {
        self.feedbacks.write().clear();
    }

    /// Remove feedback that meets specific criteria.
    pub fn remove_feedback(&self, key: &str) -> Option<FeedbackDrivenSelectivity> {
        self.feedbacks.write().remove(key)
    }

    /// Number of feedback requests received
    pub fn feedback_count(&self) -> usize {
        self.feedbacks.read().len()
    }

    /// Set default parameters
    pub fn set_default_params(&mut self, alpha: f64, min_correction: f64, max_correction: f64) {
        self.default_alpha = alpha.clamp(0.0, 1.0);
        self.default_min_correction = min_correction.max(0.01);
        self.default_max_correction = max_correction.max(self.default_min_correction);
    }
}

impl Default for SelectivityFeedbackManager {
    fn default() -> Self {
        Self::new()
    }
}

/// Whether two independent samples agree closely enough for the staged
/// one to be adopted.
fn agrees(first: f64, second: f64) -> bool {
    if first <= 0.0 || second <= 0.0 {
        return first == second;
    }
    first.max(second) / first.min(second) <= CORROBORATION_AGREEMENT
}

/// Minimum execution samples before a learned correction is served to
/// planning decisions. A single sample (possibly distorted by per-operator
/// cardinality mismatch) must not flip structural go/no-go gates; repeated
/// consistent executions promote the correction to trusted.
pub const MIN_STABLE_FEEDBACK_COUNT: u64 = 3;

/// Immutable point-in-time copy of the learned selectivity corrections.
///
/// Planning reads through a snapshot taken at the start of one decision, so
/// repeated estimation of the same predicate can neither register new state
/// nor move baselines: a decision is a pure function of the statistics plus
/// the snapshot. Execution feedback keeps writing to the live manager, which
/// later snapshots pick up as legitimate learning.
#[derive(Debug, Clone, Default)]
pub struct SelectivitySnapshot {
    corrections: HashMap<String, FeedbackDrivenSelectivity>,
}

impl SelectivitySnapshot {
    /// Correct a fresh estimate with the snapshotted factor, if the
    /// condition had learned a stable correction (enough execution samples)
    /// when the snapshot was taken. Otherwise `None`, and the caller keeps
    /// the fresh statistics estimate.
    pub fn corrected(&self, key: &str, fresh_estimate: f64) -> Option<f64> {
        self.corrections
            .get(key)
            .filter(|feedback| feedback.feedback_count() >= MIN_STABLE_FEEDBACK_COUNT)
            .map(|feedback| feedback.corrected_for_estimate(fresh_estimate))
    }

    /// Number of corrected conditions captured by the snapshot.
    pub fn len(&self) -> usize {
        self.corrections.len()
    }

    /// Whether the snapshot carries no learned corrections.
    pub fn is_empty(&self) -> bool {
        self.corrections.is_empty()
    }
}

impl Clone for SelectivityFeedbackManager {
    fn clone(&self) -> Self {
        Self {
            feedbacks: RwLock::new(self.feedbacks.read().clone()),
            default_alpha: self.default_alpha,
            default_min_correction: self.default_min_correction,
            default_max_correction: self.default_max_correction,
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_feedback_driven_selectivity() {
        let mut feedback = FeedbackDrivenSelectivity::new(0.1);
        assert_eq!(feedback.estimated_selectivity(), 0.1);

        // Updated feedback
        feedback.update_with_feedback(0.15);
        feedback.update_with_feedback(0.12);

        // The corrected selectivity should be within a reasonable range.
        let corrected = feedback.corrected_selectivity();
        assert!(corrected > 0.0 && corrected <= 0.99);

        // The number of feedback instances should be 2.
        assert_eq!(feedback.feedback_count(), 2);
    }

    #[test]
    fn test_feedback_correction_range() {
        let mut feedback = FeedbackDrivenSelectivity::new(0.5);

        // Numerous updates; testing of the correction factor limitations
        for _ in 0..100 {
            feedback.update_with_feedback(0.01); // Far lower than the estimated values
        }

        // The correction factor should be limited to its minimum value.
        assert!(feedback.correction_factor() >= 0.1);
    }

    #[test]
    fn test_estimation_confidence() {
        let mut feedback = FeedbackDrivenSelectivity::new(0.1);
        // Initial confidence should be 0.5 (sigmoid(0) = 0.5)
        let initial_confidence = feedback.estimation_confidence();
        assert!(initial_confidence < 0.55 && initial_confidence > 0.45);

        // Add multiple feedback entries.
        for i in 0..100 {
            feedback.update_with_feedback(0.1 + (i as f64 * 0.001));
        }

        assert!(feedback.estimation_confidence() > 0.9); // High confidence level after receiving the feedback
    }

    #[test]
    fn test_avg_estimation_error() {
        let mut feedback = FeedbackDrivenSelectivity::new(0.5);

        // 1.0% error without feedback
        assert_eq!(feedback.avg_estimation_error(), 1.0);

        // Add feedback
        feedback.update_with_feedback(0.5); // No errors.
        feedback.update_with_feedback(0.6); // There are errors.

        assert!(feedback.avg_estimation_error() < 1.0);
    }

    #[test]
    fn test_selectivity_feedback_manager() {
        let manager = SelectivityFeedbackManager::new();
        manager.register_condition("age > 25".to_string(), 0.3);
        manager.register_condition("salary > 5000".to_string(), 0.2);

        assert_eq!(manager.feedback_count(), 2);

        // Updated feedback
        manager.update_feedback("age > 25", 0.35);
        manager.update_feedback("salary > 5000", 0.18);

        // Obtain the corrected version of the selective content.
        let corrected_age = manager.get_corrected_selectivity("age > 25");
        assert!(corrected_age.is_some());

        // Trying to obtain a condition that does not exist…
        assert!(manager.get_corrected_selectivity("unknown").is_none());
    }

    #[test]
    fn test_selectivity_cap() {
        let mut feedback = FeedbackDrivenSelectivity::new(0.9);
        feedback.set_selectivity_cap(0.95);

        // Numerous updates have been made in an attempt to increase the level of selectivity beyond the maximum limit.
        for _ in 0..50 {
            feedback.update_with_feedback(1.0);
        }

        // The corrected selectivity should not exceed the upper limit.
        assert!(feedback.corrected_selectivity() <= 0.95);
    }

    #[test]
    fn test_measured_ratio_stages_outlier_until_corroborated() {
        let mut feedback = FeedbackDrivenSelectivity::new(0.5);
        // Establish a baseline near 1.0 with consistent samples.
        for _ in 0..3 {
            feedback.update_with_measured_ratio(1.0, 2000);
        }
        let baseline = feedback.correction_factor();
        assert!((baseline - 1.0).abs() < 1e-9);
        assert_eq!(feedback.feedback_count(), 3);

        // A lone distorted sample (e.g. a truncated partition slice)
        // must not move the factor by itself.
        feedback.update_with_measured_ratio(0.0, 2000);
        assert!((feedback.correction_factor() - baseline).abs() < 1e-9);
        assert_eq!(feedback.feedback_count(), 3);

        // An independent execution agreeing with the staged sample
        // adopts both: the factor moves and both samples count.
        feedback.update_with_measured_ratio(0.0, 2000);
        assert!(feedback.correction_factor() < baseline);
        assert_eq!(feedback.feedback_count(), 5);
    }

    #[test]
    fn test_measured_ratio_weights_step_by_input_rows() {
        let mut tiny = FeedbackDrivenSelectivity::new(0.5);
        tiny.update_with_measured_ratio(0.1, 10);
        let mut large = FeedbackDrivenSelectivity::new(0.5);
        large.update_with_measured_ratio(0.1, 100_000);
        // Same ratio, same single sample: the table-scale measurement
        // moves the factor far more than the tiny one.
        assert!(large.correction_factor() < tiny.correction_factor());
        assert_eq!(tiny.total_input_rows(), 10);
        assert_eq!(large.total_input_rows(), 100_000);
    }

    #[test]
    fn test_measured_ratio_rejects_invalid_samples() {
        let mut feedback = FeedbackDrivenSelectivity::new(0.5);
        feedback.update_with_measured_ratio(f64::NAN, 100);
        feedback.update_with_measured_ratio(-1.0, 100);
        assert_eq!(feedback.feedback_count(), 0);
        assert!((feedback.correction_factor() - 1.0).abs() < 1e-9);
    }
}