Introduce scheduler_config column to the job_configs table (ArroyoSystems#1064)

* Introduce `scheduler_config` column to the job_configs table

Adds a per-job scheduler configuration that overlays the controller's
global scheduler config at scheduling time. For the Kubernetes
scheduler, jobs can now override image, image_pull_policy,
image_pull_secrets, command, service_account_name, resources,
task_slots, env, volumes, volume_mounts, node_selector, tolerations,
labels, and annotations.

Persistence

  * Postgres migration V31 (SQLite V9) adds a nullable JSONB
    `scheduler_config` column to `job_configs`. NULL means "no
    per-job config — use the controller's global scheduler config".

Wire format

  * Payload is versioned: { "version": 1, "type": "kubernetes", ... }
  * `version` is mandatory on the wire; mirrors the `state_context`
    pattern (see V29). Unknown versions deserialize but are rejected by
    `SchedulerConfig::ensure_supported_version()`.

Validation

  * arroyo-api/src/jobs.rs::validate_scheduler_config rejects payloads
    whose version is unknown or whose variant doesn't match the
    controller's active scheduler (HTTP 400). Called early in
    create_pipeline_inner so bad payloads short-circuit before SQL
    compilation.

Read path

  * Job API responses surface scheduler_config as raw serde_json::Value
    (with #[schema(value_type = SchedulerConfig)] keeping the OpenAPI
    spec strongly typed). No silent rewriting or dropping on read.
  * Controller stores the raw JSON on JobConfig and deserializes at
    the consumer (states::Scheduling) with fatal() on failure, mirroring
    how env_vars is handled. A malformed row fails only that job's
    state machine, not the whole updater loop.

K8s merge semantics (resolved_config)

  * image, image_pull_policy, command, service_account_name,
    task_slots, resources → replace
  * labels, annotations, node_selector → merge map (per-job wins,
    intrinsic labels still applied last)
  * image_pull_secrets, volumes, volume_mounts, tolerations, env →
    append (global first, per-job last)

* Address Micah's comments

* Pass the configs as an opaque blob through the API
* Use figment for config merging

* Introduce fatal error within the scehduler

* Move is_empty_overlay to the schedulers/mod.rs file

* Fix SQLite NOT NULL constraint on scheduler_config

The scheduler_config column is NOT NULL, but the API was passing
serde_json::Value::Null (the default of serde_json::Value) when the
client omitted the field. Postgres jsonb accepts a JSON null as a
valid value, but SQLite TEXT does not distinguish between SQL NULL
and JSON null, causing the integ tests to fail with:

    NOT NULL constraint failed: job_configs.scheduler_config

Change PipelinePost.scheduler_config to Option<serde_json::Value>
(mirroring the env_vars pattern) and collapse None / Some(Null) into
an empty object at the API boundary so the column always carries a
valid JSON object.

Author: QnJ1c2kNCg

Cargo.lock

@@ -0,0 +1,4 @@
+-- Per-job scheduler config overlay. Same shape as the controller's
+-- global scheduler config (e.g. kubernetes-scheduler.*); an empty
+-- object means "no overrides, use the global config as-is".
+ALTER TABLE job_configs ADD COLUMN scheduler_config JSONB NOT NULL DEFAULT '{}'::jsonb;

crates/arroyo-api/migrations/V31__add_job_config_scheduler_config.sql

@@ -188,8 +188,8 @@ WHERE id = :job_id AND organization_id = :organization_id;
 
 --! create_job(ttl_micros?)
 INSERT INTO job_configs
-(id, organization_id, pipeline_name, created_by, pipeline_id, checkpoint_interval_micros, ttl_micros, env_vars)
-VALUES (:id, :organization_id, :pipeline_name, :created_by, :pipeline_id, :checkpoint_interval_micros, :ttl_micros, :env_vars);
+(id, organization_id, pipeline_name, created_by, pipeline_id, checkpoint_interval_micros, ttl_micros, env_vars, scheduler_config)
+VALUES (:id, :organization_id, :pipeline_name, :created_by, :pipeline_id, :checkpoint_interval_micros, :ttl_micros, :env_vars, :scheduler_config);
 
 --! create_job_status
 INSERT INTO job_statuses (pub_id, id, organization_id) VALUES (:pub_id, :id, :organization_id);
@@ -203,23 +203,23 @@ WHERE job_configs.organization_id = :organization_id AND ttl_micros IS NULL
 ORDER BY COALESCE(job_configs.updated_at, job_configs.created_at) DESC;
 
 --! get_pipeline_jobs : DbPipelineJob(start_time?, finish_time?, state?, tasks?, failure_message?, failure_domain?, run_id?, state_context?)
-SELECT job_configs.id, stop, start_time, finish_time, state, tasks, failure_message, failure_domain, run_id, checkpoint_interval_micros, job_configs.created_at, state_context
+SELECT job_configs.id, stop, start_time, finish_time, state, tasks, failure_message, failure_domain, run_id, checkpoint_interval_micros, job_configs.created_at, state_context, scheduler_config
 FROM job_configs
          INNER JOIN job_statuses ON job_configs.id = job_statuses.id
          INNER JOIN pipelines ON pipelines.id = job_configs.pipeline_id
 WHERE job_configs.organization_id = :organization_id AND pipelines.pub_id = :pub_id
 ORDER BY job_configs.created_at DESC;
 
 --! get_all_jobs : DbPipelineJob(start_time?, finish_time?, state?, tasks?, failure_message?, failure_domain?, run_id?, state_context?)
-SELECT job_configs.id, stop, start_time, finish_time, state, tasks, failure_message, failure_domain, run_id, checkpoint_interval_micros, job_configs.created_at, state_context
+SELECT job_configs.id, stop, start_time, finish_time, state, tasks, failure_message, failure_domain, run_id, checkpoint_interval_micros, job_configs.created_at, state_context, scheduler_config
 FROM job_configs
          INNER JOIN job_statuses ON job_configs.id = job_statuses.id
          INNER JOIN pipelines ON pipelines.id = job_configs.pipeline_id
 WHERE job_configs.organization_id = :organization_id AND ttl_micros IS NULL
 ORDER BY job_configs.created_at DESC;
 
 --! get_pipeline_job : DbPipelineJob(start_time?, finish_time?, state?, tasks?, failure_message?, failure_domain?, run_id?, state_context?)
-SELECT job_configs.id, stop, start_time, finish_time, state, tasks, failure_message, failure_domain, run_id, checkpoint_interval_micros, job_configs.created_at, state_context
+SELECT job_configs.id, stop, start_time, finish_time, state, tasks, failure_message, failure_domain, run_id, checkpoint_interval_micros, job_configs.created_at, state_context, scheduler_config
 FROM job_configs
          INNER JOIN job_statuses ON job_configs.id = job_statuses.id
          INNER JOIN pipelines ON pipelines.id = job_configs.pipeline_id

crates/arroyo-api/queries/api_queries.sql

@@ -0,0 +1 @@
+ALTER TABLE job_configs ADD COLUMN scheduler_config TEXT NOT NULL DEFAULT '{}';

crates/arroyo-api/sqlite_migrations/V9__add_job_config_scheduler_config.sql

@@ -113,6 +113,7 @@ fn operator_checkpoint_groups(
     operators
 }
 
+#[allow(clippy::too_many_arguments)]
 pub(crate) async fn create_job(
     pipeline_name: &str,
     pipeline_id: i64,
@@ -121,6 +122,7 @@ pub(crate) async fn create_job(
     auth: &AuthData,
     db: &DatabaseSource,
     env_vars: HashMap<String, String>,
+    scheduler_config: serde_json::Value,
 ) -> Result<String, ErrorResp> {
     let checkpoint_interval = if preview {
         Duration::from_secs(24 * 60 * 60)
@@ -177,6 +179,7 @@ pub(crate) async fn create_job(
             None
         }),
         &env_vars_json,
+        &scheduler_config,
     )
     .await?;
 

crates/arroyo-api/src/jobs.rs

@@ -306,6 +306,7 @@ pub(crate) async fn create_pipeline_int(
     state_url: Option<String>,
     tags: HashMap<String, String>,
     env_vars: HashMap<String, String>,
+    scheduler_config: serde_json::Value,
 ) -> Result<String, ErrorResp> {
     if parallelism > auth.org_metadata.max_parallelism as u64 {
         return Err(bad_request(format!(
@@ -443,6 +444,7 @@ pub(crate) async fn create_pipeline_int(
         &auth,
         db,
         env_vars,
+        scheduler_config,
     )
     .await?;
 
@@ -532,6 +534,7 @@ impl From<DbPipelineJob> for Job {
                     .unwrap_or_default(),
             }),
             created_at: to_micros(val.created_at),
+            scheduler_config: val.scheduler_config,
         }
     }
 }
@@ -671,6 +674,10 @@ async fn create_pipeline_inner(
         pipeline_post.state_url,
         pipeline_post.tags.unwrap_or_default(),
         pipeline_post.env_vars.unwrap_or_default(),
+        match pipeline_post.scheduler_config {
+            None | Some(serde_json::Value::Null) => serde_json::Value::Object(Default::default()),
+            Some(v) => v,
+        },
     )
     .await?;
 
@@ -725,6 +732,7 @@ pub async fn create_preview_pipeline(
         None,
         HashMap::default(),
         HashMap::default(),
+        serde_json::Value::Object(Default::default()),
     )
     .await?;
 

crates/arroyo-api/src/pipelines.rs

@@ -34,6 +34,8 @@ kube = { version = "0.99", features = ["runtime", "derive"] }
 k8s-openapi = { workspace = true, features = ["v1_30"] }
 shlex = "1.3"
 
+figment = { version = "0.10", features = ["json"] }
+
 # json-schema support
 serde_json = { workspace = true }
 

crates/arroyo-controller/Cargo.toml

@@ -23,7 +23,8 @@ SELECT
     restart_mode,
     ignore_state_before_epoch,
     state_context,
-    env_vars
+    env_vars,
+    scheduler_config
 FROM job_configs c
 INNER JOIN job_statuses s ON c.id = s.id;
 

crates/arroyo-controller/queries/controller_queries.sql

@@ -60,7 +60,7 @@ use types::public::{RestartMode, StopMode};
 
 pub const CHECKPOINTS_TO_KEEP: u32 = 5;
 
-#[derive(Eq, PartialEq, Clone, Debug)]
+#[derive(PartialEq, Clone, Debug)]
 pub struct JobConfig {
     id: Arc<String>,
     organization_id: String,
@@ -75,6 +75,11 @@ pub struct JobConfig {
     ignore_state_before_epoch: Option<i32>,
     /// Per-job environment variables forwarded to workers at scheduling time.
     env_vars: serde_json::Value,
+    /// Per-job scheduler configuration overlay as raw JSON (same
+    /// shape as the controller-wide scheduler config). The scheduler
+    /// interprets this; the controller treats it as opaque. An empty
+    /// object is the no-override case.
+    scheduler_config: serde_json::Value,
 }
 
 /// Per-pipeline data that doesn't change for the lifetime of a job.
@@ -628,6 +633,7 @@ impl ControllerServer {
                         restart_mode: p.restart_mode,
                         ignore_state_before_epoch: p.ignore_state_before_epoch,
                         env_vars: p.env_vars,
+                        scheduler_config: p.scheduler_config,
                     };
 
                     let mut jobs = jobs.lock().await;