[manager] Only keep relations with lateness in the schema.

ryzhyk · ryzhyk · commit d37e1808999d · 2025-10-21T16:35:05.000Z
We include program IR and schema in pipeline config. The config is
passed to the pipeline via ConfigMap, which has a size limit of 1MB. A
large program can easily exceed this limit. Previously we reduced IR
size by only keeping source and sink nodes. A program with a very large
number of relations can still exceed the limit. Since this information
is currently only used as part of backfill avoidance, which only cares
about relations with lateness, we prune all relations without lateness
to further reduce pipeline config size.

Signed-off-by: Leonid Ryzhyk &lt;ryzhyk@gmail.com&gt;
diff --git a/crates/feldera-types/src/program_schema.rs b/crates/feldera-types/src/program_schema.rs
@@ -174,7 +174,7 @@ impl ProgramSchema {
         self.inputs
             .iter()
             .chain(self.outputs.iter())
-            .filter(|rel| rel.fields.iter().any(|f| f.lateness.is_some()))
+            .filter(|rel| rel.has_lateness())
             .map(|rel| rel.name.clone())
             .collect()
     }
@@ -234,6 +234,10 @@ impl Relation {
         let name = canonical_identifier(name);
         self.fields.iter().find(|f| f.name == name)
     }
+
+    pub fn has_lateness(&self) -> bool {
+        self.fields.iter().any(|f| f.lateness.is_some())
+    }
 }
 
 /// A SQL field.
diff --git a/crates/pipeline-manager/src/db/types/program.rs b/crates/pipeline-manager/src/db/types/program.rs
@@ -754,7 +754,7 @@ pub fn generate_pipeline_config(
     // Only keep tables and views, ignoring intermediate nodes.
     // These are currently the only nodes used by the pipeline
     // (to compute pipeline diffs). Including all nodes can cause the IR
-    // to exceed the maximum ConfigMap size supported by k8s (3145728).
+    // to exceed the maximum ConfigMap size supported by k8s (1MB).
     let mir = program_info
         .dataflow
         .as_ref()
@@ -772,9 +772,19 @@ pub fn generate_pipeline_config(
         })
         .unwrap_or_default();
 
+    // Remove inputs and outputs that do not have lateness.
+    // This field is currently only used for backfill avoidance, which only cares about
+    // relations with lateness. Including the entire schema would cause the IR to exceed the
+    // maximum ConfigMap size supported by k8s (1MB).
+    let mut program_schema = program_info.schema.clone();
+    program_schema.inputs.retain(|input| input.has_lateness());
+    program_schema
+        .outputs
+        .retain(|output| output.has_lateness());
+
     let program_ir = ProgramIr {
         mir,
-        program_schema: program_info.schema.clone(),
+        program_schema,
     };
 
     PipelineConfig {

Original file line number	Diff line number	Diff line change
`@@ -174,7 +174,7 @@ impl ProgramSchema {`
`174`	`174`	`self.inputs`
`175`	`175`	`.iter()`
`176`	`176`	`.chain(self.outputs.iter())`
`177`		`- .filter(\|rel\| rel.fields.iter().any(\|f\| f.lateness.is_some()))`
	`177`	`+ .filter(\|rel\| rel.has_lateness())`
`178`	`178`	`.map(\|rel\| rel.name.clone())`
`179`	`179`	`.collect()`
`180`	`180`	`}`
`@@ -234,6 +234,10 @@ impl Relation {`
`234`	`234`	`let name = canonical_identifier(name);`
`235`	`235`	`self.fields.iter().find(\|f\| f.name == name)`
`236`	`236`	`}`
	`237`	`+`
	`238`	`+ pub fn has_lateness(&self) -> bool {`
	`239`	`+ self.fields.iter().any(\|f\| f.lateness.is_some())`
	`240`	`+ }`
`237`	`241`	`}`
`238`	`242`
`239`	`243`	`/// A SQL field.`