From 4242a1b00710060bda5be16f778bbdcb7f166661 Mon Sep 17 00:00:00 2001 From: Dnreikronos Date: Thu, 2 Jul 2026 20:54:08 -0300 Subject: [PATCH] Ignore replay-safe input connector changes --- crates/adapters/src/controller.rs | 187 ++++++++++++++---- .../adapters/src/controller/pipeline_diff.rs | 66 ++++++- crates/feldera-types/src/config.rs | 27 +++ python/tests/platform/test_bootstrapping.py | 63 ++++++ 4 files changed, 306 insertions(+), 37 deletions(-) diff --git a/crates/adapters/src/controller.rs b/crates/adapters/src/controller.rs index e90ee69795d..7f04a0aba7a 100644 --- a/crates/adapters/src/controller.rs +++ b/crates/adapters/src/controller.rs @@ -4764,20 +4764,6 @@ impl ControllerInit { ) } - // Transfer HTTP input endpoints that are not affected by the program diff from the checkpoint to the new configuration. - checkpoint_config - .inputs - .iter() - .filter(|(_connector_name, connector_config)| { - connector_config.connector_config.transport.is_http_input() - && !pipeline_diff.is_affected_relation(&connector_config.stream) - }) - .for_each(|(connector_name, connector_config)| { - config - .inputs - .insert(connector_name.clone(), connector_config.clone()); - }); - // Merge `config` (the configuration provided by the pipeline manager) // with `checkpoint_config` (the configuration read from the // checkpoint). @@ -4830,26 +4816,11 @@ impl ControllerInit { pipeline_template_configmap: config.global.pipeline_template_configmap.clone(), }, - // If pipeline is unmodified, we may need to replay journaled inputs. - // We therefore use connector configuration from the checkpoint, including - // transient HTTP and adhoc connectors, so that we can use them to - // replay journaled inputs. - inputs: if !modified { - checkpoint_config - .inputs - .into_iter() - .filter(|(_, config)| { - // The clock input connector will be automatically recreated and initialized - // with the clock resolution from the pipeline config. - !matches!( - config.connector_config.transport, - TransportConfig::ClockInput(_) - ) - }) - .collect() - } else { - config.inputs - }, + inputs: Self::inputs_for_checkpoint_replay( + config.inputs, + checkpoint_config.inputs, + &pipeline_diff, + ), outputs: if !modified { checkpoint_config.outputs } else { @@ -4880,6 +4851,69 @@ impl ControllerInit { }) } + fn inputs_for_checkpoint_replay( + mut config_inputs: BTreeMap, InputEndpointConfig>, + checkpoint_inputs: BTreeMap, InputEndpointConfig>, + pipeline_diff: &feldera_types::pipeline_diff::PipelineDiff, + ) -> BTreeMap, InputEndpointConfig> { + // Preserve the pipeline manager's input configs before HTTP input + // endpoints are transferred from the checkpoint below. The final + // replay config still needs replay-safe limits from the new config. + let new_inputs = config_inputs.clone(); + + // Transfer HTTP input endpoints that are not affected by the program + // diff from the checkpoint to the new configuration. + checkpoint_inputs + .iter() + .filter(|(_connector_name, connector_config)| { + connector_config.connector_config.transport.is_http_input() + && !pipeline_diff.is_affected_relation(&connector_config.stream) + }) + .for_each(|(connector_name, connector_config)| { + config_inputs.insert(connector_name.clone(), connector_config.clone()); + }); + + if pipeline_diff.is_empty() { + // If the pipeline is unmodified, we may need to replay journaled + // inputs. We therefore use connector configuration from the + // checkpoint, including transient HTTP and adhoc connectors, so + // that we can use them to replay journaled inputs. Input + // flow-control settings are safe to adopt from the new config + // without invalidating checkpointed state. + Self::checkpoint_inputs_with_replay_safe_config(checkpoint_inputs, &new_inputs) + } else { + // Otherwise, keep the new pipeline-manager config, including + // transferred HTTP and adhoc inputs needed to replay journaled + // inputs after a program-diff bootstrap. + config_inputs + } + } + + fn checkpoint_inputs_with_replay_safe_config( + mut checkpoint_inputs: BTreeMap, InputEndpointConfig>, + new_inputs: &BTreeMap, InputEndpointConfig>, + ) -> BTreeMap, InputEndpointConfig> { + for (connector_name, checkpoint_input) in checkpoint_inputs.iter_mut() { + if let Some(new_input) = new_inputs.get(connector_name) { + checkpoint_input + .connector_config + .apply_input_checkpoint_replay_config_from(&new_input.connector_config); + } + } + + checkpoint_inputs + .into_iter() + .filter(|(_, config)| { + // The clock input connector will be automatically recreated and initialized + // with the clock resolution from the pipeline config. + !matches!( + config.connector_config.transport, + TransportConfig::ClockInput(_) + ) + }) + .collect() + } + pub fn set_incarnation_uuid(&mut self, incarnation_uuid: Uuid) { self.incarnation_uuid = incarnation_uuid } @@ -8378,7 +8412,10 @@ mod controller_init_tests { use super::ControllerInit; use crate::ControllerError; use feldera_adapterlib::errors::controller::ConfigError; - use feldera_types::config::{PipelineConfig, RuntimeConfig}; + use feldera_types::config::{InputEndpointConfig, PipelineConfig, RuntimeConfig}; + use feldera_types::pipeline_diff::PipelineDiff; + use serde_json::json; + use std::{borrow::Cow, collections::BTreeMap}; fn pipeline_config(global: RuntimeConfig) -> PipelineConfig { PipelineConfig { @@ -8442,7 +8479,85 @@ mod controller_init_tests { if matches!( *config_error, ConfigError::DatafusionMemoryExceedsBudget { .. }, - ), + ), )); } + + #[test] + fn checkpoint_inputs_adopt_replay_safe_config() { + let mut checkpoint_inputs = BTreeMap::new(); + checkpoint_inputs.insert( + Cow::Borrowed("test_input.connector"), + input_config(json!({"name": "empty_input"}), 100), + ); + + let mut new_inputs = BTreeMap::new(); + new_inputs.insert( + Cow::Borrowed("test_input.connector"), + input_config(json!({"name": "empty_input"}), 200), + ); + + let merged = ControllerInit::checkpoint_inputs_with_replay_safe_config( + checkpoint_inputs, + &new_inputs, + ); + let connector_config = &merged["test_input.connector"].connector_config; + + assert_eq!(connector_config.max_queued_records, 200); + assert_eq!(connector_config.max_queued_bytes, Some(400)); + assert_eq!(connector_config.max_batch_size, Some(201)); + assert_eq!(connector_config.max_worker_batch_size, Some(202)); + } + + #[test] + fn checkpoint_http_input_replay_uses_pre_transfer_config() { + let mut checkpoint_inputs = BTreeMap::new(); + checkpoint_inputs.insert( + Cow::Borrowed("test_input.connector"), + input_config( + json!({ + "name": "http_input", + "config": {"name": "test_input.connector"} + }), + 100, + ), + ); + + let mut config_inputs = BTreeMap::new(); + config_inputs.insert( + Cow::Borrowed("test_input.connector"), + input_config( + json!({ + "name": "http_input", + "config": {"name": "test_input.connector"} + }), + 200, + ), + ); + + let merged = ControllerInit::inputs_for_checkpoint_replay( + config_inputs, + checkpoint_inputs, + &PipelineDiff::new_with_program_diff(Default::default()), + ); + let connector_config = &merged["test_input.connector"].connector_config; + + assert!(connector_config.transport.is_http_input()); + assert_eq!(connector_config.max_queued_records, 200); + assert_eq!(connector_config.max_queued_bytes, Some(400)); + assert_eq!(connector_config.max_batch_size, Some(201)); + assert_eq!(connector_config.max_worker_batch_size, Some(202)); + } + + fn input_config(transport: serde_json::Value, max_queued_records: u64) -> InputEndpointConfig { + serde_json::from_value(json!({ + "stream": "test_input", + "transport": transport, + "max_queued_records": max_queued_records, + "max_queued_bytes": max_queued_records * 2, + "max_batch_size": max_queued_records + 1, + "max_worker_batch_size": max_queued_records + 2, + })) + .unwrap() + } } diff --git a/crates/adapters/src/controller/pipeline_diff.rs b/crates/adapters/src/controller/pipeline_diff.rs index f9e5ead96bf..fef329127f0 100644 --- a/crates/adapters/src/controller/pipeline_diff.rs +++ b/crates/adapters/src/controller/pipeline_diff.rs @@ -147,7 +147,7 @@ pub fn compute_pipeline_diff( .get(*k) .unwrap() .connector_config - .equal_modulo_paused(&v.connector_config) + .equal_for_input_checkpoint_replay(&v.connector_config) }) .map(|(k, _)| k.to_string()) .collect::>(); @@ -197,3 +197,67 @@ pub fn compute_pipeline_diff( .with_modified_output_connectors(modified_output_connectors) .with_removed_output_connectors(removed_output_connectors)) } + +#[cfg(test)] +mod tests { + use super::compute_pipeline_diff; + use feldera_types::config::PipelineConfig; + use serde_json::{Map, Value, json}; + + fn input_endpoint(mut fields: Map) -> Value { + let mut endpoint = Map::from_iter([ + ("stream".to_string(), json!("t1")), + ("transport".to_string(), json!({"name": "empty_input"})), + ]); + endpoint.append(&mut fields); + Value::Object(endpoint) + } + + fn pipeline_config(input: Value) -> PipelineConfig { + serde_json::from_value(json!({ + "name": "test", + "workers": 1, + "inputs": { + "t1.connector": input + } + })) + .unwrap() + } + + #[test] + fn input_flow_control_changes_do_not_modify_connector() { + let old_config = pipeline_config(input_endpoint(Map::new())); + let new_config = pipeline_config(input_endpoint(Map::from_iter([ + ("max_queued_records".to_string(), json!(5000)), + ("max_queued_bytes".to_string(), json!(6000)), + ("max_batch_size".to_string(), json!(7000)), + ("max_worker_batch_size".to_string(), json!(8000)), + ("paused".to_string(), json!(true)), + ]))); + + let diff = compute_pipeline_diff(&old_config, &new_config).unwrap(); + + assert!(diff.modified_input_connectors().is_empty()); + } + + #[test] + fn input_transport_changes_still_modify_connector() { + let old_config = pipeline_config(input_endpoint(Map::new())); + let new_config = pipeline_config(input_endpoint(Map::from_iter([( + "transport".to_string(), + json!({ + "name": "datagen", + "config": { + "plan": [{"limit": 1}] + } + }), + )]))); + + let diff = compute_pipeline_diff(&old_config, &new_config).unwrap(); + + assert_eq!( + diff.modified_input_connectors(), + &vec!["t1.connector".to_string()] + ); + } +} diff --git a/crates/feldera-types/src/config.rs b/crates/feldera-types/src/config.rs index ad34a38e74d..e881be1693e 100644 --- a/crates/feldera-types/src/config.rs +++ b/crates/feldera-types/src/config.rs @@ -1723,6 +1723,33 @@ impl ConnectorConfig { a == b } + /// Compare two input connector configs modulo fields that only affect + /// runtime flow control and do not invalidate checkpointed connector state. + pub fn equal_for_input_checkpoint_replay(&self, other: &Self) -> bool { + let mut a = self.clone(); + let mut b = other.clone(); + a.normalize_for_input_checkpoint_replay(); + b.normalize_for_input_checkpoint_replay(); + a == b + } + + fn normalize_for_input_checkpoint_replay(&mut self) { + self.paused = false; + self.max_batch_size = None; + self.max_worker_batch_size = None; + self.max_queued_records = default_max_queued_records(); + self.max_queued_bytes = None; + } + + /// Adopt input connector settings that are safe to change while replaying + /// checkpointed connector state. + pub fn apply_input_checkpoint_replay_config_from(&mut self, other: &Self) { + self.max_batch_size = other.max_batch_size; + self.max_worker_batch_size = other.max_worker_batch_size; + self.max_queued_records = other.max_queued_records; + self.max_queued_bytes = other.max_queued_bytes; + } + /// Returns `max_queued_records` or, if it is not set, the default. pub fn max_queued_records(&self) -> u64 { self.max_queued_records diff --git a/python/tests/platform/test_bootstrapping.py b/python/tests/platform/test_bootstrapping.py index 84acfd3f26a..b5fd5d6154a 100644 --- a/python/tests/platform/test_bootstrapping.py +++ b/python/tests/platform/test_bootstrapping.py @@ -708,3 +708,66 @@ def gen_sql(connectors): pipeline.checkpoint(True) pipeline.stop(force=True) + + +@enterprise_only +@single_host_only +@gen_pipeline_name +def test_bootstrap_input_flow_control_changes(pipeline_name): + """ + Enterprise: input connector flow-control changes should not require bootstrapping. + """ + + def gen_sql(max_queued_records): + connectors = json.dumps( + [ + { + "name": "datagen_connector", + "transport": { + "name": "datagen", + "config": {"plan": [{"limit": 10}]}, + }, + "max_queued_records": max_queued_records, + "max_queued_bytes": max_queued_records * 100, + "max_batch_size": max_queued_records + 1, + "max_worker_batch_size": max_queued_records + 2, + } + ] + ) + return f"""CREATE TABLE t1(x int) +with ( + 'materialized' = 'true', + 'connectors' = '{connectors}' +); +CREATE MATERIALIZED VIEW v1 AS SELECT COUNT(*) AS c FROM t1; +""" + + pipeline = PipelineBuilder( + TEST_CLIENT, + pipeline_name, + sql=gen_sql(100), + runtime_config=RuntimeConfig( + workers=FELDERA_TEST_NUM_WORKERS, + hosts=FELDERA_TEST_NUM_HOSTS, + fault_tolerance_model=None, + ), + ).create_or_replace() + + pipeline.start() + assert pipeline.status() == PipelineStatus.RUNNING + + pipeline.wait_for_completion(timeout_s=300) + result = list(pipeline.query("SELECT * FROM v1;")) + assert result == [{"c": 10}] + + pipeline.checkpoint(True) + pipeline.stop(force=True) + + pipeline.modify(sql=gen_sql(200)) + pipeline.start(bootstrap_policy=BootstrapPolicy.REJECT, timeout_s=300) + assert pipeline.status() == PipelineStatus.RUNNING + + result = list(pipeline.query("SELECT * FROM v1;")) + assert result == [{"c": 10}] + + pipeline.stop(force=True)