use parking_lot::RwLock;
use std::collections::HashMap;
#[derive(Debug, Clone)]
pub struct FeedbackDrivenSelectivity {
estimated_selectivity: f64,
actual_selectivity_ewma: f64,
correction_factor: f64,
feedback_count: u64,
alpha: f64,
min_correction: f64,
max_correction: f64,
selectivity_cap: f64,
cumulative_estimation_error: f64,
error_sum_squares: f64,
total_input_rows: u64,
pending_outlier: Option<MeasuredSample>,
}
#[derive(Debug, Clone, Copy)]
struct MeasuredSample {
ratio: f64,
input_rows: u64,
}
const OUTLIER_DEVIATION: f64 = 4.0;
const CORROBORATION_AGREEMENT: f64 = 2.0;
const FULL_WEIGHT_INPUT_ROWS: f64 = 1000.0;
impl FeedbackDrivenSelectivity {
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,
}
}
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,
}
}
pub fn estimated_selectivity(&self) -> f64 {
self.estimated_selectivity
}
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;
}
}
pub fn corrected_selectivity(&self) -> f64 {
self.corrected_for_estimate(self.estimated_selectivity)
}
pub fn corrected_for_estimate(&self, estimated: f64) -> f64 {
(estimated * self.correction_factor)
.clamp(self.min_correction * estimated, self.selectivity_cap)
}
pub fn correction_factor(&self) -> f64 {
self.correction_factor
}
pub fn feedback_count(&self) -> u64 {
self.feedback_count
}
pub fn total_input_rows(&self) -> u64 {
self.total_input_rows
}
pub fn estimation_confidence(&self) -> f64 {
let x = self.feedback_count as f64 * 0.1;
1.0 / (1.0 + (-x).exp())
}
pub fn avg_estimation_error(&self) -> f64 {
if self.feedback_count == 0 {
return 1.0;
}
self.cumulative_estimation_error / self.feedback_count as f64
}
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()
}
pub fn update_with_feedback(&mut self, actual_selectivity: f64) {
if self.estimated_selectivity <= 0.0 {
return;
}
let ratio = actual_selectivity / self.estimated_selectivity;
let estimated = self.corrected_selectivity();
let error = (actual_selectivity - estimated).abs();
self.cumulative_estimation_error += error;
self.error_sum_squares += error * error;
self.correction_factor = (1.0 - self.alpha) * self.correction_factor + self.alpha * ratio;
self.correction_factor = self
.correction_factor
.clamp(self.min_correction, self.max_correction);
self.actual_selectivity_ewma =
(1.0 - self.alpha) * self.actual_selectivity_ewma + self.alpha * actual_selectivity;
self.feedback_count += 1;
}
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;
self.correction_factor = (1.0 - self.alpha) * self.correction_factor + self.alpha * 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.feedback_count += 1;
}
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);
}
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
}
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;
}
pub fn update_with_batch(&mut self, actual_selectivities: &[f64]) {
for &selectivity in actual_selectivities {
self.update_with_feedback(selectivity);
}
}
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;
}
pub fn set_alpha(&mut self, alpha: f64) {
self.alpha = alpha.clamp(0.0, 1.0);
}
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);
}
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)
}
}
#[derive(Debug)]
pub struct SelectivityFeedbackManager {
feedbacks: RwLock<HashMap<String, FeedbackDrivenSelectivity>>,
default_alpha: f64,
default_min_correction: f64,
default_max_correction: f64,
}
impl SelectivityFeedbackManager {
pub fn new() -> Self {
Self {
feedbacks: RwLock::new(HashMap::new()),
default_alpha: 0.3,
default_min_correction: 0.1,
default_max_correction: 10.0,
}
}
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,
}
}
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);
}
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
}
}
pub fn get_corrected_selectivity(&self, key: &str) -> Option<f64> {
self.feedbacks
.read()
.get(key)
.map(|f| f.corrected_selectivity())
}
pub fn snapshot(&self) -> SelectivitySnapshot {
SelectivitySnapshot {
corrections: self.feedbacks.read().clone(),
}
}
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
}
}
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
}
}
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
}
}
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
}
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);
}
}
}
pub fn get_feedback(&self, key: &str) -> Option<FeedbackDrivenSelectivity> {
self.feedbacks.read().get(key).cloned()
}
pub fn get_all_keys(&self) -> Vec<String> {
self.feedbacks.read().keys().cloned().collect()
}
pub fn clear_all(&self) {
self.feedbacks.write().clear();
}
pub fn remove_feedback(&self, key: &str) -> Option<FeedbackDrivenSelectivity> {
self.feedbacks.write().remove(key)
}
pub fn feedback_count(&self) -> usize {
self.feedbacks.read().len()
}
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()
}
}
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
}
pub const MIN_STABLE_FEEDBACK_COUNT: u64 = 3;
#[derive(Debug, Clone, Default)]
pub struct SelectivitySnapshot {
corrections: HashMap<String, FeedbackDrivenSelectivity>,
}
impl SelectivitySnapshot {
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))
}
pub fn len(&self) -> usize {
self.corrections.len()
}
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);
feedback.update_with_feedback(0.15);
feedback.update_with_feedback(0.12);
let corrected = feedback.corrected_selectivity();
assert!(corrected > 0.0 && corrected <= 0.99);
assert_eq!(feedback.feedback_count(), 2);
}
#[test]
fn test_feedback_correction_range() {
let mut feedback = FeedbackDrivenSelectivity::new(0.5);
for _ in 0..100 {
feedback.update_with_feedback(0.01); }
assert!(feedback.correction_factor() >= 0.1);
}
#[test]
fn test_estimation_confidence() {
let mut feedback = FeedbackDrivenSelectivity::new(0.1);
let initial_confidence = feedback.estimation_confidence();
assert!(initial_confidence < 0.55 && initial_confidence > 0.45);
for i in 0..100 {
feedback.update_with_feedback(0.1 + (i as f64 * 0.001));
}
assert!(feedback.estimation_confidence() > 0.9); }
#[test]
fn test_avg_estimation_error() {
let mut feedback = FeedbackDrivenSelectivity::new(0.5);
assert_eq!(feedback.avg_estimation_error(), 1.0);
feedback.update_with_feedback(0.5); feedback.update_with_feedback(0.6);
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);
manager.update_feedback("age > 25", 0.35);
manager.update_feedback("salary > 5000", 0.18);
let corrected_age = manager.get_corrected_selectivity("age > 25");
assert!(corrected_age.is_some());
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);
for _ in 0..50 {
feedback.update_with_feedback(1.0);
}
assert!(feedback.corrected_selectivity() <= 0.95);
}
#[test]
fn test_measured_ratio_stages_outlier_until_corroborated() {
let mut feedback = FeedbackDrivenSelectivity::new(0.5);
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);
feedback.update_with_measured_ratio(0.0, 2000);
assert!((feedback.correction_factor() - baseline).abs() < 1e-9);
assert_eq!(feedback.feedback_count(), 3);
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);
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);
}
}