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

fix(flow): validate retained source windows before recovery

Browse Source 
Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
This commit is contained in:
discord9
2026-09-22 19:32:29 +08:00
committed by discord9
parent bd564839b7
commit f9c2a9427a
2 changed files with 290 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/flow/src/batching_mode/task.rs
+126 -1
View File
@@ -18,13 +18,15 @@ use std::time::{Duration, SystemTime, UNIX_EPOCH};
use api::v1::{CreateTableExpr, TableName};
use catalog::CatalogManagerRef;
use catalog::kvbackend::KvBackendCatalogManager;
use client::OutputWithMetrics;
use common_error::ext::BoxedError;
use common_meta::key::schema_name::SchemaNameKey;
use common_query::OutputData;
use common_query::logical_plan::breakup_insert_plan;
use common_telemetry::tracing::warn;
use common_telemetry::{debug, info};
use common_time::Timestamp;
use common_time::{TimeToLive, Timestamp};
use datafusion::datasource::DefaultTableSource;
use datafusion::sql::unparser::expr_to_sql;
use datafusion_common::tree_node::{Transformed, TreeNode};
@@ -583,6 +585,129 @@ impl BatchingTask {
.map(Some)
}
/// Verifies that recovery can still read all source data required by its retained scope.
pub async fn validate_recovery_retention(
&self,
retention_lower: Option<Timestamp>,
windows: &[(Timestamp, Timestamp)],
) -> Result<(), Error> {
for name in &self.config.source_table_names {
let table = self
.config
.catalog_manager
.table(&name[0], &name[1], &name[2], None)
.await
.map_err(BoxedError::new)
.context(ExternalSnafu)?
.context(UnexpectedSnafu {
reason: format!(
"Flow {} source table {} is unavailable for recovery retention validation",
self.config.flow_id,
name.join(".")
),
})?;
let ttl = if let Some(ttl) = table.table_info().meta.options.ttl {
ttl
} else {
let manager = self
.config
.catalog_manager
.as_any()
.downcast_ref::<KvBackendCatalogManager>()
.context(UnexpectedSnafu {
reason: format!(
"Flow {} cannot resolve inherited TTL for source table {} during recovery",
self.config.flow_id,
name.join(".")
),
})?;
manager
.table_metadata_manager_ref()
.schema_manager()
.get(SchemaNameKey::new(&name[0], &name[1]))
.await
.map_err(BoxedError::new)
.context(ExternalSnafu)?
.context(UnexpectedSnafu {
reason: format!(
"Flow {} schema {}.{} is unavailable for recovery retention validation",
self.config.flow_id, name[0], name[1]
),
})?
.ttl
.map(Into::into)
.unwrap_or(TimeToLive::Forever)
};
match ttl {
TimeToLive::Forever => {}
TimeToLive::Instant => {
return UnexpectedSnafu {
reason: format!(
"Flow {} source table {} has instant TTL and cannot be recovered",
self.config.flow_id,
name.join(".")
),
}
.fail();
}
TimeToLive::Duration(ttl) => {
let expire_after = self.config.expire_after.context(UnexpectedSnafu {
reason: format!(
"Flow {} source table {} has TTL {ttl:?}, but recovery has no expire_after",
self.config.flow_id,
name.join(".")
),
})?;
let expire_after = u64::try_from(expire_after).map_err(|_| {
UnexpectedSnafu {
reason: format!(
"Flow {} has negative expire_after {expire_after} during recovery retention validation",
self.config.flow_id
),
}
.build()
})?;
if ttl <= Duration::from_secs(expire_after) {
return UnexpectedSnafu {
reason: format!(
"Flow {} source table {} TTL {ttl:?} must exceed expire_after {expire_after}s for recovery",
self.config.flow_id,
name.join(".")
),
}
.fail();
}
let mut oldest = retention_lower.context(UnexpectedSnafu {
reason: format!(
"Flow {} source table {} has finite TTL {ttl:?}, but recovery has no retained lower bound",
self.config.flow_id,
name.join(".")
),
})?;
for (start, _) in windows {
oldest = oldest.min(*start);
}
let cutoff = Timestamp::current_millis()
.sub_duration(ttl)
.map_err(BoxedError::new)
.context(ExternalSnafu)?;
if oldest <= cutoff {
return UnexpectedSnafu {
reason: format!(
"Flow {} recovery requires source table {} data from {oldest:?}, older than TTL {ttl:?} cutoff {cutoff:?}",
self.config.flow_id,
name.join(".")
),
}
.fail();
}
}
}
}
Ok(())
}
/// Discovers the source time windows touched by the exact sequence range `(C, H]`.
///
/// This compatibility wrapper captures the retention bound before discovery.
src/flow/src/batching_mode/task/test.rs
+164 -1
View File
@@ -17,6 +17,7 @@ use std::pin::Pin;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::task::{Context, Poll};
use std::time::Duration;
use catalog::memory::MemoryCatalogManager;
use catalog::{DeregisterTableRequest, RegisterTableRequest};
@@ -28,7 +29,7 @@ use common_error::status_code::StatusCode;
use common_query::Output;
use common_recordbatch::RecordBatch;
use common_recordbatch::adapter::{RecordBatchMetrics, RegionWatermarkEntry};
use common_time::Timestamp;
use common_time::{TimeToLive, Timestamp};
use datatypes::data_type::ConcreteDataType as CDT;
use datatypes::schema::{ColumnSchema, Schema};
use datatypes::vectors::{
@@ -819,6 +820,47 @@ fn register_auto_created_aggregate_sink(query_engine: &QueryEngineRef, table_nam
memory_catalog.register_table_sync(request).unwrap();
}
async fn configure_source_ttl(query_engine: &QueryEngineRef, ttl: Option<TimeToLive>) {
let catalog_manager = query_engine.engine_state().catalog_manager();
let source = catalog_manager
.table(
DEFAULT_CATALOG_NAME,
DEFAULT_SCHEMA_NAME,
"numbers_with_ts",
None,
)
.await
.unwrap()
.unwrap();
let mut info = (*source.table_info()).clone();
info.meta.options.ttl = ttl;
let source = Arc::new(Table::new(
Arc::new(info),
FilterPushDownType::Unsupported,
source.data_source(),
));
let memory_catalog = catalog_manager
.as_any()
.downcast_ref::<MemoryCatalogManager>()
.unwrap();
memory_catalog
.deregister_table_sync(DeregisterTableRequest {
catalog: DEFAULT_CATALOG_NAME.to_string(),
schema: DEFAULT_SCHEMA_NAME.to_string(),
table_name: "numbers_with_ts".to_string(),
})
.unwrap();
memory_catalog
.register_table_sync(RegisterTableRequest {
catalog: DEFAULT_CATALOG_NAME.to_string(),
schema: DEFAULT_SCHEMA_NAME.to_string(),
table_name: "numbers_with_ts".to_string(),
table_id: source.table_info().table_id(),
table: source,
})
.unwrap();
}
async fn configure_source_capability(
query_engine: &QueryEngineRef,
engine: &str,
@@ -875,6 +917,127 @@ async fn configure_source_capability(
.unwrap();
}
#[tokio::test]
async fn test_validate_recovery_retention_explicit_source_ttls() {
struct Case {
name: &'static str,
ttl: Option<TimeToLive>,
expire_after: Option<i64>,
retention_age: Option<Duration>,
window_age: Option<Duration>,
should_pass: bool,
}
let cases = [
Case {
name: "finite TTL exceeding expire_after is admissible",
ttl: Some(TimeToLive::Duration(Duration::from_secs(2 * 60 * 60))),
expire_after: Some(60 * 60),
retention_age: Some(Duration::from_secs(10 * 60)),
window_age: None,
should_pass: true,
},
Case {
name: "TTL equal to expire_after is rejected",
ttl: Some(TimeToLive::Duration(Duration::from_secs(60 * 60))),
expire_after: Some(60 * 60),
retention_age: Some(Duration::from_secs(10 * 60)),
window_age: None,
should_pass: false,
},
Case {
name: "TTL shorter than expire_after is rejected",
ttl: Some(TimeToLive::Duration(Duration::from_secs(59 * 60))),
expire_after: Some(60 * 60),
retention_age: Some(Duration::from_secs(10 * 60)),
window_age: None,
should_pass: false,
},
Case {
name: "aligned window older than TTL is rejected",
ttl: Some(TimeToLive::Duration(Duration::from_secs(2 * 60 * 60))),
expire_after: Some(60 * 60),
retention_age: Some(Duration::from_secs(10 * 60)),
window_age: Some(Duration::from_secs(3 * 60 * 60)),
should_pass: false,
},
Case {
name: "old retention cutoff is rejected even without windows",
ttl: Some(TimeToLive::Duration(Duration::from_secs(2 * 60 * 60))),
expire_after: Some(60 * 60),
retention_age: Some(Duration::from_secs(3 * 60 * 60)),
window_age: None,
should_pass: false,
},
Case {
name: "explicit forever is admissible",
ttl: Some(TimeToLive::Forever),
expire_after: None,
retention_age: None,
window_age: None,
should_pass: true,
},
Case {
name: "instant TTL is rejected",
ttl: Some(TimeToLive::Instant),
expire_after: Some(60 * 60),
retention_age: Some(Duration::from_secs(10 * 60)),
window_age: None,
should_pass: false,
},
];
for case in cases {
let TestTaskParts {
mut task,
query_engine,
..
} = new_test_task_engine_and_plan_with_query(
"SELECT number, ts FROM numbers_with_ts",
"sink",
)
.await;
Arc::get_mut(&mut task.config).unwrap().expire_after = case.expire_after;
configure_source_ttl(&query_engine, case.ttl).await;
let now = Timestamp::current_millis();
let retention_lower = case.retention_age.map(|age| now.sub_duration(age).unwrap());
let windows = case
.window_age
.map(|age| {
let start = now.sub_duration(age).unwrap();
vec![(start, now)]
})
.unwrap_or_default();
assert_eq!(
task.validate_recovery_retention(retention_lower, &windows)
.await
.is_ok(),
case.should_pass,
"{}",
case.name
);
}
}
#[tokio::test]
async fn test_validate_recovery_retention_rejects_unknown_inherited_ttl() {
let TestTaskParts {
mut task,
query_engine,
..
} = new_test_task_engine_and_plan_with_query("SELECT number, ts FROM numbers_with_ts", "sink")
.await;
Arc::get_mut(&mut task.config).unwrap().expire_after = Some(60 * 60);
configure_source_ttl(&query_engine, None).await;
assert!(
task.validate_recovery_retention(Some(Timestamp::current_millis()), &[],)
.await
.is_err()
);
}
fn dirty_marker() -> DirtyTimeWindows {
let mut dirty = DirtyTimeWindows::default();
dirty.set_dirty();