rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-10-12 15:02:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(flow): freeze explicit recovery windows and retention bounds

Browse Source 
Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
This commit is contained in:
discord9
2026-09-22 19:22:32 +08:00
committed by discord9
parent 3573c55187
commit bd564839b7
3 changed files with 245 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/flow/src/batching_mode/state.rs
+128 -11
View File
@@ -208,6 +208,25 @@ impl TaskState {
self.pending_fenced_repair.as_ref()
}
/// Start repairing explicit bounded windows under a frozen high `H`.
///
/// The caller supplies the complete repair scope. Live dirty windows are
/// left unchanged so signals received after `H` remain separate.
pub fn start_fenced_repair_windows(
&mut self,
high: BTreeMap<u64, u64>,
windows: Vec<(Timestamp, Timestamp)>,
) {
let mut pending_windows = self.dirty_time_windows.clone();
pending_windows.clean();
pending_windows.add_windows(windows);
self.pending_fenced_repair = Some(FencedRepair {
high,
pending_windows,
});
self.checkpoint_mode = CheckpointMode::FullSnapshot;
}
/// Finish the fenced repair and promote the frozen high watermark to the
/// checkpoint map. Incremental-disabled flows stay in FullSnapshot mode.
pub fn finish_fenced_repair(&mut self) -> Option<BTreeMap<u64, u64>> {
@@ -259,22 +278,18 @@ impl TaskState {
task_ctx: Option<&BatchingTask>,
) -> Result<Option<FilterExprInfo>, Error> {
if let Some(repair) = self.pending_fenced_repair.as_mut() {
let expr = repair.pending_windows.gen_filter_exprs(
// Fenced windows are an explicit frozen repair scope. They must not
// be pruned by the moving live-data expiration boundary, and an
// empty repair must remain active until its high watermark is
// explicitly finished.
return repair.pending_windows.gen_filter_exprs(
col_name,
expire_lower_bound,
None,
window_size,
window_cnt,
flow_id,
task_ctx,
)?;
if expr.is_some() || !repair.pending_windows.is_empty() {
return Ok(expr);
}
// All pending repair windows may have expired during merge. Clear
// the empty repair so this call can fall back to live dirty windows
// instead of routing future executions to an empty queue forever.
self.pending_fenced_repair = None;
);
}
self.dirty_time_windows.gen_filter_exprs(
@@ -1393,6 +1408,108 @@ mod test {
assert_eq!(state.dirty_time_windows.len(), 2);
}
#[test]
fn test_explicit_fenced_repair_windows_keep_live_windows_separate() {
let mut state = state_with_past_update(Duration::from_secs(1));
state
.dirty_time_windows
.add_window(Timestamp::new_second(0), Some(Timestamp::new_second(1_000)));
let high = BTreeMap::from([(1, 10)]);
state.start_fenced_repair_windows(
high.clone(),
vec![(Timestamp::new_second(10), Timestamp::new_second(15))],
);
assert_eq!(state.checkpoint_mode(), CheckpointMode::FullSnapshot);
assert_eq!(state.pending_fenced_repair().unwrap().high(), &high);
assert_eq!(
state
.pending_fenced_repair()
.unwrap()
.pending_windows()
.len(),
1
);
assert_eq!(state.dirty_time_windows.len(), 1);
let filter = state
.gen_scoped_filter_exprs(
"ts",
Some(Timestamp::new_second(100)),
chrono::Duration::seconds(5),
1,
1,
None,
)
.unwrap()
.unwrap();
assert_eq!(
filter.time_ranges,
vec![(Timestamp::new_second(10), Timestamp::new_second(15))]
);
state
.dirty_time_windows
.add_window(Timestamp::new_second(20), Some(Timestamp::new_second(25)));
state.restore_scoped_windows(&filter);
assert_eq!(
state
.pending_fenced_repair()
.unwrap()
.pending_windows()
.len(),
1
);
assert_eq!(state.dirty_time_windows.len(), 2);
}
#[test]
fn test_explicit_fenced_repair_keeps_empty_high_and_old_windows() {
let mut state = state_with_past_update(Duration::from_secs(1));
state
.dirty_time_windows
.add_window(Timestamp::new_second(0), Some(Timestamp::new_second(1_000)));
let high = BTreeMap::from([(1, 10)]);
state.start_fenced_repair_windows(high.clone(), Vec::new());
assert!(
state
.gen_scoped_filter_exprs(
"ts",
Some(Timestamp::new_second(100)),
chrono::Duration::seconds(5),
1,
1,
None,
)
.unwrap()
.is_none()
);
assert_eq!(state.pending_fenced_repair().unwrap().high(), &high);
assert_eq!(state.dirty_time_windows.len(), 1);
assert_eq!(state.finish_fenced_repair(), Some(high));
state.start_fenced_repair_windows(
BTreeMap::from([(1, 11)]),
vec![(Timestamp::new_second(-20), Timestamp::new_second(-15))],
);
let filter = state
.gen_scoped_filter_exprs(
"ts",
Some(Timestamp::new_second(100)),
chrono::Duration::seconds(5),
1,
1,
None,
)
.unwrap()
.unwrap();
assert_eq!(
filter.time_ranges,
vec![(Timestamp::new_second(-20), Timestamp::new_second(-15))]
);
}
#[test]
fn test_disable_incremental_persists_full_snapshot_mode() {
let query_ctx = QueryContext::arc();
src/flow/src/batching_mode/task.rs
+64 -33
View File
@@ -539,16 +539,75 @@ impl BatchingTask {
Arc::new(query_ctx.fork())
}
/// Returns the retention lower bound aligned to this task's time window.
pub fn recovery_retention_lower_bound(&self) -> Result<Option<Timestamp>, Error> {
let Some(expire_after) = self.config.expire_after else {
return Ok(None);
};
let expire_after = u64::try_from(expire_after).map_err(|_| {
UnexpectedSnafu {
reason: format!(
"Flow {} has negative expire_after {expire_after}",
self.config.flow_id
),
}
.build()
})?;
let now = Timestamp::new_second(
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map_err(|err| {
UnexpectedSnafu {
reason: format!("Failed to read recovery wall clock: {err}"),
}
.build()
})?
.as_secs() as i64,
);
let lower = now
.sub_duration(Duration::from_secs(expire_after))
.map_err(BoxedError::new)
.context(ExternalSnafu)?;
self.config
.time_window_expr
.as_ref()
.context(UnexpectedSnafu {
reason: "Recovery expiry requires a time-window expression".to_string(),
})?
.eval(lower)?
.0
.context(UnexpectedSnafu {
reason: "Recovery expiry time-window expression returned no lower bound"
.to_string(),
})
.map(Some)
}
/// Discovers the source time windows touched by the exact sequence range `(C, H]`.
///
/// Terminal proof must cover ALL regions in `C`; subset proofs and pruning are rejected.
/// The caller owns the execution guard. This method is read-only: it executes a
/// timestamp projection of the aggregate input and leaves task state untouched.
/// This compatibility wrapper captures the retention bound before discovery.
pub async fn capture_recovery_windows(
&self,
engine: &QueryEngineRef,
frontend_client: &FrontendClient,
lower: &BTreeMap<u64, u64>,
) -> Result<(BTreeMap<u64, u64>, Vec<(Timestamp, Timestamp)>), Error> {
let retention_lower = self.recovery_retention_lower_bound()?;
self.capture_recovery_windows_since(engine, frontend_client, lower, retention_lower)
.await
}
/// Discovers windows under `lower` using a caller-frozen retention bound.
///
/// The caller owns the execution guard and freezes this bound with its recovery scope.
/// This read-only method leaves task state untouched. Terminal proof must cover every
/// region in `lower`; subset or pruned proofs are rejected.
pub async fn capture_recovery_windows_since(
&self,
engine: &QueryEngineRef,
frontend_client: &FrontendClient,
lower: &BTreeMap<u64, u64>,
retention_lower: Option<Timestamp>,
) -> Result<(BTreeMap<u64, u64>, Vec<(Timestamp, Timestamp)>), Error> {
if lower.is_empty() {
return UnexpectedSnafu {
@@ -600,37 +659,9 @@ impl BatchingTask {
self.config.flow_id
),
})?;
let input = if let Some(expire_after) = self.config.expire_after {
let expire_after = u64::try_from(expire_after).map_err(|_| {
UnexpectedSnafu {
reason: format!(
"Flow {} has negative expire_after {expire_after}",
self.config.flow_id
),
}
.build()
})?;
let now = Timestamp::new_second(
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map_err(|err| {
UnexpectedSnafu {
reason: format!("Failed to read recovery wall clock: {err}"),
}
.build()
})?
.as_secs() as i64,
);
let lower = now
.sub_duration(Duration::from_secs(expire_after))
.map_err(BoxedError::new)
.context(ExternalSnafu)?;
let lower = time_window_expr.eval(lower)?.0.context(UnexpectedSnafu {
reason: "Recovery expiry time-window expression returned no lower bound"
.to_string(),
})?;
let input = if let Some(retention_lower) = retention_lower {
let mut add_filter = AddFilterRewriter::new(
col(&time_window_expr.column_name).gt_eq(lit(to_df_literal(lower)?)),
col(&time_window_expr.column_name).gt_eq(lit(to_df_literal(retention_lower)?)),
);
input
.rewrite(&mut add_filter)
src/flow/src/batching_mode/task/test.rs
+53 -5
View File
@@ -1749,6 +1749,47 @@ async fn test_capture_recovery_windows_decodes_raw_timestamp_projection_with_whe
assert_recovery_state_unchanged(&task, &before_checkpoints, &before_dirty);
}
#[tokio::test]
async fn test_capture_recovery_windows_since_uses_supplied_retention_lower() {
let TestTaskParts {
task,
query_engine,
..
} = new_time_window_test_task_with_query(
"SELECT max(number) AS output_value, date_bin(INTERVAL '5 second', ts) AS output_window FROM numbers_with_ts WHERE number = 42 GROUP BY output_window",
)
.await;
configure_source_capability(&query_engine, "mito", true).await;
let lower = BTreeMap::from([(1, 10)]);
let retention_lower = Timestamp::new_second(-10);
let handler = Arc::new(RecoveryCaptureHandler {
output: std::sync::Mutex::new(Some(recovery_stream_output(
vec![],
Some(vec![(1, Some(10))]),
false,
))),
lower: serde_json::to_string(&lower).unwrap(),
expire_after: Some(1),
expiry_lower: std::sync::Mutex::new(None),
});
let frontend = FrontendClient::from_grpc_handler(
Arc::downgrade(
&(handler.clone()
as Arc<dyn crate::batching_mode::frontend_client::GrpcQueryHandlerWithBoxedError>),
),
QueryOptions::default(),
);
let (high, windows) = task
.capture_recovery_windows_since(&query_engine, &frontend, &lower, Some(retention_lower))
.await
.unwrap();
assert_eq!(high, lower);
assert!(windows.is_empty());
assert_eq!(*handler.expiry_lower.lock().unwrap(), Some(retention_lower));
}
#[tokio::test]
async fn test_capture_recovery_windows_terminal_proof_cases_leave_state_unchanged() {
struct Case {
@@ -2915,7 +2956,7 @@ async fn test_full_snapshot_scoped_plan_marks_checkpoint_advance_safe_only_after
}
#[tokio::test]
async fn test_expired_empty_fenced_repair_generates_scoped_base_repair_plan() {
async fn test_expired_fenced_repair_uses_frozen_scope() {
let TestTaskParts {
mut task,
query_engine,
@@ -2953,11 +2994,18 @@ async fn test_expired_empty_fenced_repair_generates_scoped_base_repair_plan() {
)
.await
.unwrap()
.expect("expired empty repair should fall back to live dirty");
.unwrap();
assert!(matches!(plan.coverage, QueryCoverage::ScopedBaseRepair));
assert!(plan.coverage.snapshot_seqs().is_empty());
assert!(task.state.read().unwrap().pending_fenced_repair().is_none());
// A fenced repair is a frozen explicit scope: expiry cannot discard its
// old window or make this query consume the live post-fence signal.
assert!(matches!(
plan.coverage,
QueryCoverage::FencedRepairChunk { .. }
));
let state = task.state.read().unwrap();
assert!(state.pending_fenced_repair().is_some());
assert!(state.fenced_repair_pending_is_empty());
assert_eq!(state.dirty_time_windows.len(), 1);
}
#[tokio::test]