Add _schema_validation_progress system table (#40312)

Adds a new system table, `_schema_validation_progress`, for each component. We're making this a separate table from the `_schemas` table in each component to avoid conflicts, since almost all writes depend on the `_schemas` table and the progress will updated pretty frequently during schema validation. Progress isn't 100% accurate because the document counts might be stale, but this is pretty much the same way we track index backfill progress.

GitOrigin-RevId: a4fe10c47d04e329352b48da7eab424c06ec2380

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 = 121; // nipunn
+pub const DATABASE_VERSION: DatabaseVersion = 122; // emma
 
 pub struct MigrationExecutor<RT: Runtime> {
     pub db: Database<RT>,
@@ -76,6 +76,11 @@ impl<RT: Runtime> MigrationExecutor<RT> {
                     .await?;
                 MigrationCompletionCriterion::MigrationComplete(to_version)
             },
+            122 => {
+                // This is an empty migration because we added a new system
+                // table for each component, _schema_validation_progress
+                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,11 @@ use crate::{
     exports::ExportsTable,
     external_packages::EXTERNAL_PACKAGES_TABLE,
     log_sinks::LOG_SINKS_TABLE,
+    schema_validation_progress::{
+        SchemaValidationProgressTable,
+        SCHEMA_VALIDATION_PROGRESS_BY_SCHEMA_ID,
+        SCHEMA_VALIDATION_PROGRESS_TABLE,
+    },
 };
 
 pub mod airbyte_import;
@@ -222,6 +227,7 @@ mod metrics;
 pub mod migrations;
 pub mod modules;
 pub mod scheduled_jobs;
+pub mod schema_validation_progress;
 pub mod session_requests;
 pub mod snapshot_imports;
 pub mod source_packages;
@@ -265,9 +271,10 @@ enum DefaultTableNumber {
     CanonicalUrls = 34,
     CronNextRun = 35,
     IndexBackfills = 36,
+    SchemaValidationProgress = 37,
     // 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 - 37 - emma
+    // Next Number - 38 - emma
 }
 
 impl From<DefaultTableNumber> for TableNumber {
@@ -310,6 +317,7 @@ impl From<DefaultTableNumber> for &'static dyn ErasedSystemTable {
             DefaultTableNumber::CanonicalUrls => &CanonicalUrlsTable,
             DefaultTableNumber::CronNextRun => &CronNextRunTable,
             DefaultTableNumber::IndexBackfills => &IndexBackfillTable,
+            DefaultTableNumber::SchemaValidationProgress => &SchemaValidationProgressTable,
         }
     }
 }
@@ -557,6 +565,7 @@ pub fn component_system_tables() -> Vec<&'static dyn ErasedSystemTable> {
         &ModulesTable,
         &UdfConfigTable,
         &SourcePackagesTable,
+        &SchemaValidationProgressTable,
     ]
 }
 
@@ -626,6 +635,7 @@ pub static FIRST_SEEN_TABLE: LazyLock<BTreeMap<TableName, DatabaseVersion>> = La
         FUNCTION_HANDLES_TABLE.clone() => 102,
         CANONICAL_URLS_TABLE.clone() => 116,
         INDEX_BACKFILLS_TABLE.clone() => 120,
+        SCHEMA_VALIDATION_PROGRESS_TABLE.clone() => 122,
     }
 });
 
@@ -652,6 +662,7 @@ pub static FIRST_SEEN_INDEX: LazyLock<BTreeMap<IndexName, DatabaseVersion>> = La
         BY_COMPONENT_PATH_INDEX.name() => 102,
         EXPORTS_BY_REQUESTOR.name() => 110,
         INDEX_BACKFILLS_BY_INDEX_ID.name() => 120,
+        SCHEMA_VALIDATION_PROGRESS_BY_SCHEMA_ID.name() => 122,
     }
 });
 

crates/model/src/schema_validation_progress/mod.rs

@@ -0,0 +1,46 @@
+pub mod types;
+
+use std::sync::LazyLock;
+
+use common::document::CREATION_TIME_FIELD_PATH;
+use database::system_tables::{
+    SystemIndex,
+    SystemTable,
+};
+use value::{
+    FieldPath,
+    TableName,
+};
+
+pub static SCHEMA_VALIDATION_PROGRESS_TABLE: LazyLock<TableName> = LazyLock::new(|| {
+    "_schema_validation_progress"
+        .parse()
+        .expect("Invalid built-in _schema_validation_progress table")
+});
+
+pub static SCHEMA_VALIDATION_PROGRESS_BY_SCHEMA_ID: LazyLock<
+    SystemIndex<SchemaValidationProgressTable>,
+> = LazyLock::new(|| {
+    SystemIndex::new(
+        "by_schema_id",
+        [&SCHEMA_ID_FIELD, &CREATION_TIME_FIELD_PATH],
+    )
+    .unwrap()
+});
+
+static SCHEMA_ID_FIELD: LazyLock<FieldPath> =
+    LazyLock::new(|| "schemaId".parse().expect("invalid schemaId field"));
+
+pub struct SchemaValidationProgressTable;
+
+impl SystemTable for SchemaValidationProgressTable {
+    type Metadata = types::SchemaValidationProgressMetadata;
+
+    fn table_name() -> &'static TableName {
+        &SCHEMA_VALIDATION_PROGRESS_TABLE
+    }
+
+    fn indexes() -> Vec<SystemIndex<Self>> {
+        vec![SCHEMA_VALIDATION_PROGRESS_BY_SCHEMA_ID.clone()]
+    }
+}

crates/model/src/schema_validation_progress/types.rs

@@ -0,0 +1,58 @@
+use serde::{
+    Deserialize,
+    Serialize,
+};
+use value::{
+    codegen_convex_serialization,
+    DeveloperDocumentId,
+};
+
+#[derive(Clone, Debug, Eq, PartialEq)]
+#[cfg_attr(any(test, feature = "testing"), derive(proptest_derive::Arbitrary))]
+pub struct SchemaValidationProgressMetadata {
+    /// The ID of the schema being validated. Should correspond to a document in
+    /// the _schemas table in `Pending` state.
+    pub schema_id: DeveloperDocumentId,
+    /// The number of documents that have been validated so far.
+    pub num_docs_validated: u64,
+    /// The number of total documents that need to be validated. Note this is
+    /// approximate because there could be changes since the time we wrote this
+    /// value from the table summary when the schema is submitted as pending.
+    /// It's possible for num_docs_validated to exceed total_docs.
+    pub total_docs: u64,
+}
+
+#[derive(Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct SerializedSchemaValidationProgressMetadata {
+    pub schema_id: String,
+    pub num_docs_validated: i64,
+    pub total_docs: i64,
+}
+
+impl From<SchemaValidationProgressMetadata> for SerializedSchemaValidationProgressMetadata {
+    fn from(metadata: SchemaValidationProgressMetadata) -> Self {
+        SerializedSchemaValidationProgressMetadata {
+            schema_id: metadata.schema_id.to_string(),
+            num_docs_validated: metadata.num_docs_validated as i64,
+            total_docs: metadata.total_docs as i64,
+        }
+    }
+}
+
+impl TryFrom<SerializedSchemaValidationProgressMetadata> for SchemaValidationProgressMetadata {
+    type Error = anyhow::Error;
+
+    fn try_from(serialized: SerializedSchemaValidationProgressMetadata) -> anyhow::Result<Self> {
+        Ok(SchemaValidationProgressMetadata {
+            schema_id: serialized.schema_id.parse()?,
+            num_docs_validated: serialized.num_docs_validated as u64,
+            total_docs: serialized.total_docs as u64,
+        })
+    }
+}
+
+codegen_convex_serialization!(
+    SchemaValidationProgressMetadata,
+    SerializedSchemaValidationProgressMetadata
+);