Skip to content

pipeline-manager: require program_info_url and schema unit test #6294

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ryzhyk merged 1 commit into from
May 21, 2026
Merged

pipeline-manager: require program_info_url and schema unit test #6294

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 27 additions & 2 deletions crates/pipeline-manager/src/compiler/sql_compiler.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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;
Expand All @@ -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.
Expand Down
12 changes: 6 additions & 6 deletions crates/pipeline-manager/src/db/types/program.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<String, PropertyValue>,
pub properties: BTreeMap<String, PropertyValue>,
}

#[derive(Debug, Deserialize)]
struct ProgramSchemaPropertiesOnly {
pub struct ProgramSchemaPropertiesOnly {
#[serde(default)]
inputs: Vec<RelationPropertiesOnly>,
pub inputs: Vec<RelationPropertiesOnly>,
#[serde(default)]
outputs: Vec<RelationPropertiesOnly>,
pub outputs: Vec<RelationPropertiesOnly>,
}

/// Generates the program info using the program schema.
Expand Down
12 changes: 12 additions & 0 deletions crates/pipeline-manager/src/runner/error.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,9 @@ pub enum RunnerError {
AutomatonCannotConstructProgramBinaryUrl {
error: String,
},
AutomatonCannotConstructProgramInfoUrl {
error: String,
},
AutomatonMissingDeploymentId,
AutomatonMissingDeploymentConfig,
AutomatonMissingDeploymentLocation,
Expand Down Expand Up @@ -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")
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand Down
73 changes: 36 additions & 37 deletions crates/pipeline-manager/src/runner/local_runner.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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
//
Expand Down
51 changes: 30 additions & 21 deletions crates/pipeline-manager/src/runner/pipeline_automata.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -935,7 +935,7 @@ impl<T: PipelineExecutor> PipelineAutomaton<T> {
}
};

// 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 {
Expand Down Expand Up @@ -1268,27 +1268,36 @@ impl<T: PipelineExecutor> PipelineAutomaton<T> {
},
);

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())
Expand All @@ -1315,7 +1324,7 @@ impl<T: PipelineExecutor> PipelineAutomaton<T> {
&deployment_config,
program_info,
&program_binary_url,
program_info_url.as_deref(),
&program_info_url,
pipeline.program_version,
)
.await
Expand Down Expand Up @@ -1864,7 +1873,7 @@ mod test {
_: &PipelineConfig,
_: &serde_json::Value,
_: &str,
_: Option<&str>,
_: &str,
_: Version,
) -> Result<(), ManagerError> {
Ok(())
Expand Down
2 changes: 1 addition & 1 deletion crates/pipeline-manager/src/runner/pipeline_executor.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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>;

Expand Down
Loading