1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
use super::OptimizerEngine;
use crate::optimizer::stats::feedback::query::OperatorFeedback;
impl OptimizerEngine {
/// Fold execution feedback into selectivity/cardinality corrections.
///
/// Consumes actual row counts from `feedback_history`, which execution
/// populates via `collect_execution_feedback` from runtime operator
/// profiles (estimated vs actual per shape/predicate). Observability
/// portraits (`QueryProfile` in metrics) and optimizer corrections
/// (`QueryExecutionFeedback` here) serve separate roles — human-facing
/// diagnostics vs EWMA estimate correction — and are both fed from the
/// same runtime profiles.
///
/// Each execution is folded exactly once (per-fingerprint sequence
/// watermark): re-applying the retained history on every optimization
/// would walk the corrections over identical samples again. Samples
/// from starved operators or without a measurable input pairing are
/// dropped instead of folded.
pub fn maybe_apply_feedback(&self) {
if !self.enable_feedback {
return;
}
let history = self.feedback_history();
let mut applied = 0usize;
let mut skipped_truncated = 0usize;
let mut skipped_unpaired = 0usize;
let mut decision_runs = 0usize;
for fingerprint in history.get_all_fingerprints() {
let Some(avg_error) = history.get_avg_row_error(&fingerprint) else {
continue;
};
if !self.feedback_trigger.should_trigger(avg_error) {
continue;
}
let watermark = self
.feedback_applied_seq
.lock()
.get(&fingerprint)
.copied()
.unwrap_or(0);
let mut max_seq = watermark;
let mut folded_runs = 0usize;
for feedback in history.get_feedback_for_query(&fingerprint) {
if feedback.sequence <= watermark {
continue;
}
max_seq = max_seq.max(feedback.sequence);
if feedback.apply_rows > 0 {
self.decision_feedback.record_apply_run(
space_of(&feedback),
feedback.apply_rows,
feedback.apply_time_us,
);
decision_runs += 1;
}
if feedback.join_rows > 0 {
self.decision_feedback.record_join_run(
space_of(&feedback),
feedback.join_rows,
feedback.join_time_us,
);
decision_runs += 1;
}
for op in &feedback.operator_feedbacks {
if !op.exhausted {
// Starved by Limit / early termination: partial
// counts describe the cutoff, not the operator.
skipped_truncated += 1;
continue;
}
if let Some(key) = &op.condition_key {
match self.measured_selectivity_ratio(key, op) {
Some((ratio, input_rows)) => {
if self
.selectivity_feedback
.update_measured_ratio(key, ratio, input_rows)
{
applied += 1;
}
}
None => {
skipped_unpaired += 1;
}
}
} else if let Some(key) = &op.shape_key {
let ratio = op.actual_rows as f64 / op.estimated_rows.max(1) as f64;
if self.cardinality_feedback.update_feedback_ratio(key, ratio) {
applied += 1;
}
}
}
self.feedback_trigger.record_feedback();
folded_runs += 1;
}
if folded_runs > 0 {
self.feedback_applied_seq
.lock()
.insert(fingerprint, max_seq);
self.feedback_trigger.mark_updated();
}
}
if applied > 0 {
log::debug!(
"Feedback loop: applied {} selectivity/cardinality corrections from {} fingerprints ({} truncated, {} unpaired)",
applied,
history.query_count(),
skipped_truncated,
skipped_unpaired,
);
}
if decision_runs > 0 {
log::debug!(
"Feedback loop: folded {} Apply/Join decision runs from {} fingerprints",
decision_runs,
history.query_count()
);
}
}
/// Isolate the operator-local selectivity error for a filter sample.
///
/// Returns the observed-vs-estimated selectivity ratio plus the measured
/// input rows. `None` when the pairing is unavailable (unknown key,
/// non-positive baseline, or missing/empty input measurement): such a
/// sample cannot separate predicate error from upstream cardinality
/// error and must not train the correction.
fn measured_selectivity_ratio(&self, key: &str, op: &OperatorFeedback) -> Option<(f64, u64)> {
let baseline = self
.selectivity_feedback
.get_feedback(key)?
.estimated_selectivity();
if baseline <= 0.0 {
return None;
}
let actual_in = op.actual_input_rows?;
if actual_in == 0 {
return None;
}
let observed = op.actual_rows as f64 / actual_in as f64;
Some((observed / baseline, actual_in))
}
/// Drop all feedback corrections scoped to `space` (`None` clears all).
///
/// Called when a space's statistics are invalidated (ANALYZE force or
/// DDL commit) so stale corrections cannot mislead estimates after a
/// schema or data change.
pub fn invalidate_space_feedback(&self, space: Option<&str>) {
let removed = match space {
Some(space) => {
let removed = self
.selectivity_feedback
.remove_feedback_by_space(&format!("{}:", space));
removed
+ self
.cardinality_feedback
.remove_feedback_by_space(&format!("{}:", space))
}
None => {
let keys = self.selectivity_feedback.get_all_keys();
for key in &keys {
self.selectivity_feedback.remove_feedback(key);
}
let cardinality_keys = self.cardinality_feedback.get_all_keys();
for key in &cardinality_keys {
self.cardinality_feedback.remove_feedback(key);
}
keys.len() + cardinality_keys.len()
}
};
self.decision_feedback.invalidate_space(space);
if removed > 0 {
log::info!(
"Invalidated {} feedback corrections for space {:?}",
removed,
space
);
}
}
}
/// Target space of a collected execution, defaulting to unscoped.
fn space_of(feedback: &crate::optimizer::stats::feedback::query::QueryExecutionFeedback) -> &str {
feedback.space.as_deref().unwrap_or("")
}