Support an optional created_at on pipeline creation

PipelinePost takes an optional created_at in microseconds since the Unix
epoch, applied to both the pipeline and its initial job. When omitted the
inserts fall back to CURRENT_TIMESTAMP via COALESCE, preserving the
existing database-clock behavior.

Author: QnJ1c2kNCg

crates/arroyo-api/queries/api_queries.sql

@@ -123,9 +123,9 @@ WHERE organization_id = :organization_id AND pub_id = :pub_id;
 
 --: DbPipeline (state?, ttl_micros?, state_url?)
 
---! create_pipeline(textual_repr?, state_url?)
-INSERT INTO pipelines (pub_id, organization_id, created_by, name, type, textual_repr, udfs, program, proto_version, state_url, tags)
-VALUES (:pub_id, :organization_id, :created_by, :name, :type, :textual_repr, :udfs, :program, :proto_version, :state_url, :tags);
+--! create_pipeline(textual_repr?, state_url?, created_at?)
+INSERT INTO pipelines (pub_id, organization_id, created_by, name, type, textual_repr, udfs, program, proto_version, state_url, tags, created_at)
+VALUES (:pub_id, :organization_id, :created_by, :name, :type, :textual_repr, :udfs, :program, :proto_version, :state_url, :tags, COALESCE(:created_at, CURRENT_TIMESTAMP));
 
 --! get_pipelines : DbPipeline
 SELECT pipelines.id, pipelines.pub_id, name, type, textual_repr, udfs, program, checkpoint_interval_micros, stop, pipelines.created_at, state, parallelism_overrides, ttl_micros, state_url, tags
@@ -189,10 +189,10 @@ SET
    ignore_state_before_epoch = :ignore_state_before_epoch
 WHERE id = :job_id AND organization_id = :organization_id;
 
---! create_job(ttl_micros?)
+--! create_job(ttl_micros?, created_at?)
 INSERT INTO job_configs
-(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);
+(id, organization_id, pipeline_name, created_by, pipeline_id, checkpoint_interval_micros, ttl_micros, env_vars, scheduler_config, created_at)
+VALUES (:id, :organization_id, :pipeline_name, :created_by, :pipeline_id, :checkpoint_interval_micros, :ttl_micros, :env_vars, :scheduler_config, COALESCE(:created_at, CURRENT_TIMESTAMP));
 
 --! create_job_status
 INSERT INTO job_statuses (pub_id, id, organization_id) VALUES (:pub_id, :id, :organization_id);

crates/arroyo-api/src/jobs.rs

@@ -18,6 +18,7 @@ use futures_util::stream::Stream;
 use std::convert::Infallible;
 use std::str::FromStr;
 use std::{collections::HashMap, time::Duration};
+use time::OffsetDateTime;
 use tokio_stream::StreamExt as _;
 use tokio_stream::wrappers::ReceiverStream;
 use tonic::{Code, Request};
@@ -123,6 +124,7 @@ pub(crate) async fn create_job(
     db: &DatabaseSource,
     env_vars: HashMap<String, String>,
     scheduler_config: serde_json::Value,
+    created_at: Option<OffsetDateTime>,
 ) -> Result<String, ErrorResp> {
     let checkpoint_interval = if preview {
         Duration::from_secs(24 * 60 * 60)
@@ -180,6 +182,7 @@ pub(crate) async fn create_job(
         }),
         &env_vars_json,
         &scheduler_config,
+        &created_at,
     )
     .await?;
 

crates/arroyo-api/src/lib.rs

@@ -38,7 +38,7 @@ use crate::pipelines::{
     __path_restart_pipeline, __path_validate_query,
 };
 use crate::rest::__path_ping;
-use crate::rest_utils::{ErrorResp, service_unavailable};
+use crate::rest_utils::{ErrorResp, bad_request, service_unavailable};
 use crate::udfs::{__path_create_udf, __path_delete_udf, __path_get_udfs, __path_validate_udf};
 use arroyo_rpc::api_types::{checkpoints::*, connections::*, metrics::*, pipelines::*, udfs::*, *};
 use arroyo_rpc::config::{ApiAuthMode, config};
@@ -119,6 +119,11 @@ pub(crate) fn to_micros(dt: OffsetDateTime) -> u64 {
     (dt.unix_timestamp_nanos() / 1_000) as u64
 }
 
+pub(crate) fn from_micros(micros: u64) -> Result<OffsetDateTime, ErrorResp> {
+    OffsetDateTime::from_unix_timestamp_nanos((micros as i128) * 1_000)
+        .map_err(|_| bad_request(format!("timestamp out of range: {micros}")))
+}
+
 pub async fn compiler_service() -> Result<CompilerGrpcClient<Channel>, ErrorResp> {
     // TODO: cache this
     let config = config();

crates/arroyo-api/src/pipelines.rs

@@ -51,7 +51,7 @@ use crate::rest_utils::{
 };
 use crate::types::public::{PipelineType, RestartMode, StopMode};
 use crate::udfs::build_udf;
-use crate::{connection_tables, to_micros};
+use crate::{connection_tables, from_micros, to_micros};
 use arroyo_rpc::config::{JobControllerMode, config};
 use arroyo_rpc::errors::ErrorDomain;
 use arroyo_types::to_millis;
@@ -307,6 +307,7 @@ pub(crate) async fn create_pipeline_int(
     tags: HashMap<String, String>,
     env_vars: HashMap<String, String>,
     scheduler_config: serde_json::Value,
+    created_at: Option<OffsetDateTime>,
 ) -> Result<String, ErrorResp> {
     if parallelism > auth.org_metadata.max_parallelism as u64 {
         return Err(bad_request(format!(
@@ -413,6 +414,7 @@ pub(crate) async fn create_pipeline_int(
         &2,
         &state_url,
         &tags_json,
+        &created_at,
     )
     .await?;
 
@@ -445,6 +447,7 @@ pub(crate) async fn create_pipeline_int(
         db,
         env_vars,
         scheduler_config,
+        created_at,
     )
     .await?;
 
@@ -652,6 +655,8 @@ async fn create_pipeline_inner(
 
     let udfs = pipeline_post.udfs.unwrap_or_default();
 
+    let created_at = pipeline_post.created_at.map(from_micros).transpose()?;
+
     let compiled = compile_sql(
         pipeline_post.query.clone(),
         &udfs,
@@ -681,6 +686,7 @@ async fn create_pipeline_inner(
             None | Some(serde_json::Value::Null) => serde_json::Value::Object(Default::default()),
             Some(v) => v,
         },
+        created_at,
     )
     .await?;
 
@@ -736,6 +742,7 @@ pub async fn create_preview_pipeline(
         HashMap::default(),
         HashMap::default(),
         serde_json::Value::Object(Default::default()),
+        None,
     )
     .await?;
 

crates/arroyo-rpc/src/api_types/pipelines.rs

@@ -39,6 +39,10 @@ pub struct PipelinePost {
     /// all mean "use the controller's global scheduler config
     /// unchanged".
     pub scheduler_config: Option<serde_json::Value>,
+    /// Optional creation timestamp, in microseconds since the Unix epoch,
+    /// applied to both the pipeline and its initial job. If omitted, the
+    /// database's current time is used.
+    pub created_at: Option<u64>,
 }
 
 #[derive(Serialize, Deserialize, Clone, Debug, ToSchema)]

webui/src/gen/api-types.ts

@@ -861,13 +861,31 @@ export interface components {
         PipelinePatch: {
             /** Format: int64 */
             checkpoint_interval_micros?: number | null;
+            /** @description Per-job environment variables forwarded to workers. */
+            env_vars?: {
+                [key: string]: string;
+            } | null;
             /** Format: int64 */
             parallelism?: number | null;
+            /** @description Per-job scheduler configuration overlay. The shape mirrors the
+             *     controller's global scheduler config (e.g. the
+             *     `kubernetes-scheduler.*` block) and is merged on top of it at
+             *     scheduling time. An omitted field, `null`, or an empty object
+             *     all mean "use the controller's global scheduler config
+             *     unchanged". */
+            scheduler_config?: unknown;
             stop?: components["schemas"]["StopType"] | null;
         };
         PipelinePost: {
             /** Format: int64 */
             checkpoint_interval_micros?: number | null;
+            /**
+             * Format: int64
+             * @description Optional creation timestamp, in microseconds since the Unix epoch,
+             *     applied to both the pipeline and its initial job. If omitted, the
+             *     database's current time is used.
+             */
+            created_at?: number | null;
             /** @description Per-job environment variables forwarded to workers. */
             env_vars?: {
                 [key: string]: string;