diff --git a/src/pipeline/orchestrator.rs b/src/pipeline/orchestrator.rs index 9393c22..573f5cc 100644 --- a/src/pipeline/orchestrator.rs +++ b/src/pipeline/orchestrator.rs @@ -572,6 +572,7 @@ impl Orchestrator { let mut pre_warm = tokio::spawn(pre_warm_schemas( Arc::clone(&http_client), collect_pre_warm_tables(&self.config.routing), + stages.default_table.clone(), Duration::from_secs(self.config.schema.pre_warm_retry_secs), schema_result_tx.clone(), self.shutdown.clone(), @@ -654,6 +655,7 @@ impl Orchestrator { pre_warm = tokio::spawn(pre_warm_schemas( Arc::clone(&http_client), to_warm, + stages.default_table.clone(), Duration::from_secs(new_config.schema.pre_warm_retry_secs), schema_result_tx.clone(), self.shutdown.clone(), @@ -2678,6 +2680,7 @@ fn directives_from( async fn pre_warm_schemas( client: Arc, tables: Vec, + default_table: String, budget: Duration, results: mpsc::Sender, shutdown: CancellationToken, @@ -2686,8 +2689,9 @@ async fn pre_warm_schemas( let report = { let client = client.as_ref(); let results = &results; + let default_table = default_table.as_str(); pre_warm_with_retry( - move |t| warm_tables_once(t, client, results), + move |t| warm_tables_once(t, client, results, default_table), tables, budget, &shutdown, @@ -2715,16 +2719,27 @@ async fn pre_warm_schemas( } } -/// One pre-warm round: resolve each of `tables` and send every answer to -/// `results`. A table `ClickHouse` answered for is done, absent included, and -/// only a failed fetch is tried again. +/// Whether a pre-warm answer goes to the event loop. An absent source table is +/// held back, because the engine writes a source's routing and creates its +/// table together, and an absence recorded in that gap sends the source's rows +/// to the default table until `ABSENT_TABLE_TTL` runs out. Its first batch +/// resolves it instead. The default table is never recorded absent, so its +/// answer goes through and a missing one is warned about at once. +fn pre_warm_forwards(table: &str, schema: &SchemaResolution, default_table: &str) -> bool { + table == default_table || !matches!(schema, SchemaResolution::TableNotFound) +} + +/// One pre-warm round: resolve each of `tables` and send every answer +/// [`pre_warm_forwards`] passes to `results`. A table `ClickHouse` answered +/// for is done, absent included, and only a failed fetch is tried again. /// -/// Both references come from the caller's scope, not a closure environment, +/// The references come from the caller's scope, not a closure environment, /// which keeps `pre_warm_with_retry`'s closure bound satisfiable. async fn warm_tables_once( tables: Vec, client: &ClickHouseQueryClient, results: &mpsc::Sender, + default_table: &str, ) -> Vec<(String, bool)> { use futures::future::join_all; @@ -2737,8 +2752,12 @@ async fn warm_tables_once( // wrong-protocol config behind a bare failed-count (#115). warn!(table = %table, error = %e, "Schema pre-warm fetch failed"); } - // A closed channel means the event loop has stopped, and the task is aborted with it. - let _ = results.send(result).await; + if pre_warm_forwards(&table, &result.schema, default_table) { + // A closed channel means the event loop has stopped, and the task is aborted with it. + let _ = results.send(result).await; + } else { + debug!(table = %table, "Pre-warm found no table yet, its first batch resolves it"); + } out.push((table, fault.is_none())); } out @@ -4195,6 +4214,27 @@ mod tests { resolver.await.unwrap(); } + /// A pre-warm records no absence for a source table: one the engine + /// creates just after the routing reload would otherwise lose its rows to + /// the default table. A failed fetch, and a missing default table, still + /// reach the event loop. + #[test] + fn pre_warm_holds_back_only_an_absent_source_table() { + let default = "dfe.main"; + for (table, schema, forwarded) in [ + ("dfe.bgp-updates", SchemaResolution::TableNotFound, false), + ("dfe.bgp-updates", SchemaResolution::Unavailable, true), + (default, SchemaResolution::TableNotFound, true), + (default, SchemaResolution::Unavailable, true), + ] { + assert_eq!( + pre_warm_forwards(table, &schema, default), + forwarded, + "{table} forwarded should be {forwarded}" + ); + } + } + /// A table that does not exist leaves no directive or computed-column /// entry behind, however many such names arrive. #[test]