Add _scheduled_job_args system table (#42976)

Add a new system table for storing scheduled job arguments with an index on job id.

GitOrigin-RevId: 67da6cdabced9c277b6b2b050a35d6ee3c72e562

Author: emmaling27

crates/migrations_model/src/lib.rs

@@ -30,7 +30,7 @@ pub type DatabaseVersion = i64;
 // migrations unless explicitly dropping support.
 // Add a user name next to the version when you make a change to highlight merge
 // conflicts.
-pub const DATABASE_VERSION: DatabaseVersion = 122; // emma
+pub const DATABASE_VERSION: DatabaseVersion = 123; // emma
 
 pub struct MigrationExecutor<RT: Runtime> {
     pub db: Database<RT>,
@@ -81,6 +81,11 @@ impl<RT: Runtime> MigrationExecutor<RT> {
                 // table for each component, _schema_validation_progress
                 MigrationCompletionCriterion::MigrationComplete(to_version)
             },
+            123 => {
+                // This is an empty migration because we added a new system
+                // table for each component, _scheduled_job_args
+                MigrationCompletionCriterion::MigrationComplete(to_version)
+            },
             // NOTE: Make sure to increase DATABASE_VERSION when adding new migrations.
             _ => anyhow::bail!("Version did not define a migration! {}", to_version),
         };

crates/model/src/lib.rs

@@ -199,6 +199,10 @@ use crate::{
     exports::ExportsTable,
     external_packages::EXTERNAL_PACKAGES_TABLE,
     log_sinks::LOG_SINKS_TABLE,
+    scheduled_jobs::args::{
+        ScheduledJobArgsTable,
+        SCHEDULED_JOBS_ARGS_TABLE,
+    },
 };
 
 pub mod airbyte_import;
@@ -266,9 +270,10 @@ enum DefaultTableNumber {
     CronNextRun = 35,
     IndexBackfills = 36,
     SchemaValidationProgress = 37,
+    ScheduledJobArgs = 38,
     // Keep this number and your user name up to date. The number makes it easy to know
     // what to use next. The username on the same line detects merge conflicts
-    // Next Number - 38 - emma
+    // Next Number - 39 - emma
 }
 
 impl From<DefaultTableNumber> for TableNumber {
@@ -312,6 +317,7 @@ impl From<DefaultTableNumber> for &'static dyn ErasedSystemTable {
             DefaultTableNumber::CronNextRun => &CronNextRunTable,
             DefaultTableNumber::IndexBackfills => &IndexBackfillTable,
             DefaultTableNumber::SchemaValidationProgress => &SchemaValidationProgressTable,
+            DefaultTableNumber::ScheduledJobArgs => &ScheduledJobArgsTable,
         }
     }
 }
@@ -553,6 +559,7 @@ pub fn component_system_tables() -> Vec<&'static dyn ErasedSystemTable> {
     vec![
         &FileStorageTable,
         &ScheduledJobsTable,
+        &ScheduledJobArgsTable,
         &CronJobsTable,
         &CronJobLogsTable,
         &CronNextRunTable,
@@ -629,6 +636,7 @@ pub static FIRST_SEEN_TABLE: LazyLock<BTreeMap<TableName, DatabaseVersion>> = La
         CANONICAL_URLS_TABLE.clone() => 116,
         INDEX_BACKFILLS_TABLE.clone() => 120,
         SCHEMA_VALIDATION_PROGRESS_TABLE.clone() => 122,
+        SCHEDULED_JOBS_ARGS_TABLE.clone() => 123,
     }
 });
 

crates/model/src/scheduled_jobs/args.rs

@@ -0,0 +1,29 @@
+use std::sync::LazyLock;
+
+use database::system_tables::{
+    SystemIndex,
+    SystemTable,
+};
+use value::TableName;
+
+use crate::scheduled_jobs::types::ScheduledJobArgs;
+
+pub static SCHEDULED_JOBS_ARGS_TABLE: LazyLock<TableName> = LazyLock::new(|| {
+    "_scheduled_job_args"
+        .parse()
+        .expect("_scheduled_job_args is not a valid system table name")
+});
+
+pub struct ScheduledJobArgsTable;
+
+impl SystemTable for ScheduledJobArgsTable {
+    type Metadata = ScheduledJobArgs;
+
+    fn table_name() -> &'static TableName {
+        &SCHEDULED_JOBS_ARGS_TABLE
+    }
+
+    fn indexes() -> Vec<SystemIndex<Self>> {
+        vec![]
+    }
+}

crates/model/src/scheduled_jobs/mod.rs

@@ -69,6 +69,7 @@ use crate::{
     SystemTable,
 };
 
+pub mod args;
 pub mod types;
 pub mod virtual_table;
 

crates/model/src/scheduled_jobs/types.rs

@@ -270,3 +270,38 @@ mod state {
 
     codegen_convex_serialization!(ScheduledJobState, SerializedScheduledJobState);
 }
+
+#[derive(Debug, Clone, PartialEq)]
+#[cfg_attr(any(test, feature = "testing"), derive(proptest_derive::Arbitrary))]
+pub struct ScheduledJobArgs {
+    #[cfg_attr(
+        any(test, feature = "testing"),
+        proptest(
+            strategy = "proptest::arbitrary::any_with::<ConvexArray>((0..4).into()).\
+                        prop_map(args_to_bytes).prop_filter_map(\"invalid json\", |b| b.ok())"
+        )
+    )]
+    args: ByteBuf,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct SerializedScheduledJobArgs {
+    args: ByteBuf,
+}
+
+impl From<ScheduledJobArgs> for SerializedScheduledJobArgs {
+    fn from(value: ScheduledJobArgs) -> SerializedScheduledJobArgs {
+        SerializedScheduledJobArgs { args: value.args }
+    }
+}
+
+impl TryFrom<SerializedScheduledJobArgs> for ScheduledJobArgs {
+    type Error = anyhow::Error;
+
+    fn try_from(value: SerializedScheduledJobArgs) -> anyhow::Result<Self> {
+        Ok(ScheduledJobArgs { args: value.args })
+    }
+}
+
+codegen_convex_serialization!(ScheduledJobArgs, SerializedScheduledJobArgs);