From fe1adf8283ef200c0e7a639038fffc4104366da0 Mon Sep 17 00:00:00 2001 From: Simon Kassing Date: Thu, 21 May 2026 15:24:29 +0200 Subject: [PATCH] pipeline-manager: require `program_info_url` and schema unit test The `program_info_url` is now a mandatory parameter for a pipeline to be able to be started, as its absence has been long enough deprecated. An error is added in case it is missing. This makes it consistent how program info is handled across runners. Add a unit test for the schema with only properties. Signed-off-by: Simon Kassing --- .../src/compiler/sql_compiler.rs | 29 +++++++- .../pipeline-manager/src/db/types/program.rs | 12 +-- crates/pipeline-manager/src/runner/error.rs | 12 +++ .../src/runner/local_runner.rs | 73 +++++++++---------- .../src/runner/pipeline_automata.rs | 51 +++++++------ .../src/runner/pipeline_executor.rs | 2 +- 6 files changed, 112 insertions(+), 67 deletions(-) diff --git a/crates/pipeline-manager/src/compiler/sql_compiler.rs b/crates/pipeline-manager/src/compiler/sql_compiler.rs index 31e4c3f628b..63ece844d98 100644 --- a/crates/pipeline-manager/src/compiler/sql_compiler.rs +++ b/crates/pipeline-manager/src/compiler/sql_compiler.rs @@ -911,7 +911,7 @@ mod test { use crate::auth::TenantRecord; use crate::compiler::test::{list_content_as_sorted_names, CompilerTest}; use crate::compiler::util::{create_new_file, recreate_dir}; - use crate::db::types::program::ProgramStatus; + use crate::db::types::program::{ProgramSchemaPropertiesOnly, ProgramStatus}; use crate::db::types::utils::validate_program_info; use crate::db::types::version::Version; use feldera_types::config::TransportConfig; @@ -1138,6 +1138,20 @@ mod test { assert!(subfields[1].columntype.nullable); } + // Program schema only with properties + let program_schema_properties_only: ProgramSchemaPropertiesOnly = + serde_json::from_value(program_info.schema.clone()).unwrap(); + + // Table + let table_properties_only = program_schema_properties_only.inputs.first().unwrap(); + assert_eq!(table_properties_only.name, table.name); + assert_eq!(table_properties_only.properties, table.properties); + + // View + let view_properties_only = program_schema_properties_only.outputs.get(1).unwrap(); + assert_eq!(view_properties_only.name, view.name); + assert_eq!(view_properties_only.properties, view.properties); + // Clean up test.delete_pipeline(tenant_id, pipeline_id, "p1").await; test.sql_compiler_tick().await; @@ -1232,7 +1246,7 @@ mod test { let pipeline_descr = test .check_outcome_sql_compiled(tenant_id, pipeline_id, program_code) .await; - let input_connectors = validate_program_info(&pipeline_descr.program_info.unwrap()) + let input_connectors = validate_program_info(&pipeline_descr.program_info.clone().unwrap()) .unwrap() .clone() .input_connectors; @@ -1246,6 +1260,17 @@ mod test { connector_config.transport, TransportConfig::Datagen(_) )); + + // Program schema only with properties: check properties + let program_schema_properties_only: ProgramSchemaPropertiesOnly = + serde_json::from_value(pipeline_descr.program_info.unwrap()["schema"].clone()).unwrap(); + let table_properties_only = program_schema_properties_only.inputs.first().unwrap(); + assert_eq!(table_properties_only.name, "t1"); + assert_eq!(table_properties_only.properties.len(), 1); + assert!(table_properties_only.properties.contains_key("connectors")); + assert!(table_properties_only.properties["connectors"] + .value + .contains("\"name\": \"c1\"")); } /// Tests that SQL compiler recovers from an incorrect platform version. diff --git a/crates/pipeline-manager/src/db/types/program.rs b/crates/pipeline-manager/src/db/types/program.rs index fbdd16c4c5e..4d887429e77 100644 --- a/crates/pipeline-manager/src/db/types/program.rs +++ b/crates/pipeline-manager/src/db/types/program.rs @@ -673,19 +673,19 @@ impl ProgramInfo { /// This is used to avoid parsing the entire `Relation` object, including /// SQL schema, which can change across runtime versions. #[derive(Debug, Deserialize)] -struct RelationPropertiesOnly { +pub struct RelationPropertiesOnly { #[serde(flatten)] - name: SqlIdentifier, + pub name: SqlIdentifier, #[serde(default)] - properties: BTreeMap, + pub properties: BTreeMap, } #[derive(Debug, Deserialize)] -struct ProgramSchemaPropertiesOnly { +pub struct ProgramSchemaPropertiesOnly { #[serde(default)] - inputs: Vec, + pub inputs: Vec, #[serde(default)] - outputs: Vec, + pub outputs: Vec, } /// Generates the program info using the program schema. diff --git a/crates/pipeline-manager/src/runner/error.rs b/crates/pipeline-manager/src/runner/error.rs index ab240f67922..6ae220a0591 100644 --- a/crates/pipeline-manager/src/runner/error.rs +++ b/crates/pipeline-manager/src/runner/error.rs @@ -19,6 +19,9 @@ pub enum RunnerError { AutomatonCannotConstructProgramBinaryUrl { error: String, }, + AutomatonCannotConstructProgramInfoUrl { + error: String, + }, AutomatonMissingDeploymentId, AutomatonMissingDeploymentConfig, AutomatonMissingDeploymentLocation, @@ -116,6 +119,9 @@ impl DetailedError for RunnerError { RunnerError::AutomatonCannotConstructProgramBinaryUrl { .. } => { Cow::from("AutomatonCannotConstructProgramBinaryUrl") } + RunnerError::AutomatonCannotConstructProgramInfoUrl { .. } => { + Cow::from("AutomatonCannotConstructProgramInfoUrl") + } RunnerError::AutomatonMissingDeploymentId => Cow::from("AutomatonMissingDeploymentId"), RunnerError::AutomatonMissingDeploymentConfig => { Cow::from("AutomatonMissingDeploymentConfig") @@ -196,6 +202,9 @@ impl Display for RunnerError { Self::AutomatonCannotConstructProgramBinaryUrl { error } => { write!(f, "Cannot construct program binary URL due to: {error}") } + Self::AutomatonCannotConstructProgramInfoUrl { error } => { + write!(f, "Cannot construct program info URL due to: {error}") + } Self::AutomatonMissingDeploymentId => { write!( f, @@ -402,6 +411,9 @@ impl ResponseError for RunnerError { Self::AutomatonCannotConstructProgramBinaryUrl { .. } => { StatusCode::INTERNAL_SERVER_ERROR } + Self::AutomatonCannotConstructProgramInfoUrl { .. } => { + StatusCode::INTERNAL_SERVER_ERROR + } Self::AutomatonMissingDeploymentId => StatusCode::INTERNAL_SERVER_ERROR, Self::AutomatonMissingDeploymentConfig => StatusCode::INTERNAL_SERVER_ERROR, Self::AutomatonMissingDeploymentLocation => StatusCode::INTERNAL_SERVER_ERROR, diff --git a/crates/pipeline-manager/src/runner/local_runner.rs b/crates/pipeline-manager/src/runner/local_runner.rs index d6ea3b8226d..a9b468e00ab 100644 --- a/crates/pipeline-manager/src/runner/local_runner.rs +++ b/crates/pipeline-manager/src/runner/local_runner.rs @@ -429,7 +429,7 @@ impl PipelineExecutor for LocalRunner { deployment_config: &PipelineConfig, _program_info: &serde_json::Value, program_binary_url: &str, - program_info_url: Option<&str>, + program_info_url: &str, program_version: Version, ) -> Result<(), ManagerError> { // Local runner does not support multihost. @@ -474,47 +474,46 @@ impl PipelineExecutor for LocalRunner { // Going forward, we should be able to pass program info and deployment config files // as separate arguments to the pipeline instead of merging them into one JSON file. let mut deployment_config = deployment_config.clone(); - if let Some(program_info_url) = program_info_url { - // Retrieve and store executable in pipeline working directory - let program_info_file_path = self.config.program_info_file_path(self.pipeline_id); - - self.retrieve_pipeline_file( - program_info_url, - "program info", - &program_info_file_path, - 0o660, // User: rw, Group: rw, Others: / - ) - .await?; - - // Read and parse the program info file - let program_info_contents = - fs::read_to_string(&program_info_file_path) - .await - .map_err(|e| { - ManagerError::from(CommonError::io_error( - format!( - "read program info file '{}'", - program_info_file_path.display() - ), - e, - )) - })?; - let program_info: PipelineConfigProgramInfo = - serde_json::from_str(&program_info_contents).map_err(|e| { - ManagerError::from(RunnerError::RunnerProvisionError { - error: format!( - "failed to parse program info file '{}': {e}", + // Retrieve and store program info in pipeline working directory + let program_info_file_path = self.config.program_info_file_path(self.pipeline_id); + + self.retrieve_pipeline_file( + program_info_url, + "program info", + &program_info_file_path, + 0o660, // User: rw, Group: rw, Others: / + ) + .await?; + + // Read and parse the program info file + let program_info_contents = + fs::read_to_string(&program_info_file_path) + .await + .map_err(|e| { + ManagerError::from(CommonError::io_error( + format!( + "read program info file '{}'", program_info_file_path.display() ), - }) + e, + )) })?; - // Merge program info into deployment_config - deployment_config.inputs = program_info.inputs; - deployment_config.outputs = program_info.outputs; - deployment_config.program_ir = program_info.program_ir; - } + let program_info: PipelineConfigProgramInfo = serde_json::from_str(&program_info_contents) + .map_err(|e| { + ManagerError::from(RunnerError::RunnerProvisionError { + error: format!( + "failed to parse program info file '{}': {e}", + program_info_file_path.display() + ), + }) + })?; + + // Merge program info into deployment_config + deployment_config.inputs = program_info.inputs; + deployment_config.outputs = program_info.outputs; + deployment_config.program_ir = program_info.program_ir; // Write config as YAML and JSON // diff --git a/crates/pipeline-manager/src/runner/pipeline_automata.rs b/crates/pipeline-manager/src/runner/pipeline_automata.rs index 2a36375e2d6..7788c64df81 100644 --- a/crates/pipeline-manager/src/runner/pipeline_automata.rs +++ b/crates/pipeline-manager/src/runner/pipeline_automata.rs @@ -935,7 +935,7 @@ impl PipelineAutomaton { } }; - // Input and output connectors from required program_info + // Validate the required program_info which includes input and output connectors let _program_info = match &pipeline.program_info { None => { return Action::TransitionToStopping { @@ -1268,27 +1268,36 @@ impl PipelineAutomaton { }, ); - let program_info_url = if let Some(program_info_integrity_checksum) = + let Some(program_info_integrity_checksum) = pipeline.program_info_integrity_checksum.as_ref() - { - Some(format!( - "{}://{}:{}/program_info/{}/{}/{}/{}", - if self.common_config.enable_https { - "https" - } else { - "http" - }, - self.common_config.compiler_host, - self.common_config.compiler_port, - self.pipeline_id, - pipeline.program_version, - source_checksum, - program_info_integrity_checksum, - )) - } else { - None + else { + return Action::TransitionToStopping { + error: Some( + RunnerError::AutomatonCannotConstructProgramInfoUrl { + error: "integrity checksum is missing".to_string(), + } + .into(), + ), + storage_status_details: None, + }; }; + // URL where the program info can be downloaded from + let program_info_url = format!( + "{}://{}:{}/program_info/{}/{}/{}/{}", + if self.common_config.enable_https { + "https" + } else { + "http" + }, + self.common_config.compiler_host, + self.common_config.compiler_port, + self.pipeline_id, + pipeline.program_version, + source_checksum, + program_info_integrity_checksum, + ); + let bootstrap_policy = if Self::platform_version_requires_bootstrap_policy(&pipeline.platform_version) { Some(pipeline.bootstrap_policy.unwrap_or_default()) @@ -1315,7 +1324,7 @@ impl PipelineAutomaton { &deployment_config, program_info, &program_binary_url, - program_info_url.as_deref(), + &program_info_url, pipeline.program_version, ) .await @@ -1864,7 +1873,7 @@ mod test { _: &PipelineConfig, _: &serde_json::Value, _: &str, - _: Option<&str>, + _: &str, _: Version, ) -> Result<(), ManagerError> { Ok(()) diff --git a/crates/pipeline-manager/src/runner/pipeline_executor.rs b/crates/pipeline-manager/src/runner/pipeline_executor.rs index 02e9529bfbc..3478cc98665 100644 --- a/crates/pipeline-manager/src/runner/pipeline_executor.rs +++ b/crates/pipeline-manager/src/runner/pipeline_executor.rs @@ -60,7 +60,7 @@ pub trait PipelineExecutor: Sync + Send { deployment_config: &PipelineConfig, program_info: &serde_json::Value, program_binary_url: &str, - program_info_url: Option<&str>, + program_info_url: &str, program_version: Version, ) -> Result<(), ManagerError>;