ops-catalog: thread through collection generation id on inferred schemas

This builds on previous work, which added the id of the publication that
initially created a collection as a suffix of the journal template.
Originally, this was intended to allow correct handling of cloud storage so
that a collection (or task) can be deleted and re-created again with the same
name.  The journal template suffix will be different, so that data from the
previous generation is kept separate from data for the new generation. But a
problem remains with inferred schemas, since the inferred schemas table only
stores the collection name and not the journal template suffix.

This begins formalizing that journal template suffix as a `generation_id`, and
updates the runtime and ops catalog to thread through the generation id as part
of inferred schema updates. It is materialized as part of the
`inferred_schemas` table.

This commit stops short of actually changing how inferred schema resolution works.
Notably, the primary key of the `inferred_schemas` table is unchanged, meaning that
multiple different inferred schemas still cannot coexist for the same `collection_name`.
Instead, the `collection_generation_id` is simply materialized alongside the schema,
and we can figure later how we want that to factor into inferred schema resolutions
as part of publications.

Author: psFried

Cargo.lock

@@ -35,6 +35,7 @@ proto-grpc = { path = "../proto-grpc", features = [
     "runtime_server",
 ] }
 simd-doc = { path = "../simd-doc" }
+tables = { path = "../tables" }
 tuple = { path = "../tuple" }
 unseal = { path = "../unseal" }
 

crates/runtime/Cargo.toml

@@ -30,6 +30,8 @@ pub struct Task {
 struct Binding {
     // Target collection.
     collection_name: String,
+    // Generation id of the collection, which must be output as part of updating inferred schemas.
+    collection_generation_id: models::Id,
     // JSON pointer at which document UUIDs are added.
     document_uuid_ptr: doc::Pointer,
     // Key components which are extracted from written documents.

crates/runtime/src/capture/mod.rs

@@ -332,11 +332,17 @@ pub async fn recv_client_start_commit(
     // produce structured logs of all inferred schemas that have changed
     // in this transaction.
     for binding in txn.updated_inferences.iter() {
-        let serialized = doc::shape::schema::to_schema(shapes[*binding].clone());
+        let mut serialized = doc::shape::schema::to_schema(shapes[*binding].clone());
+        let gen_id = task.bindings[*binding].collection_generation_id.to_string();
+        serialized.schema.extensions.insert(
+            "x-collection-generation-id".to_string(),
+            serde_json::Value::String(gen_id),
+        );
 
         tracing::info!(
             schema = ?ops::DebugJson(serialized),
             collection_name = %task.bindings[*binding].collection_name,
+            collection_generation_id = %task.bindings[*binding].collection_generation_id,
             binding = binding,
             "inferred schema updated"
         );

crates/runtime/src/capture/protocol.rs

@@ -132,6 +132,12 @@ impl Binding {
             state_key: _,
         } = spec;
 
+        let collection_spec = collection.as_ref().context("missing collection")?;
+        // Determine the generation id of the collection, which we'll need when updating inferred schemas.
+        // For legacy collections that don't have a generation id, we'll use the zero id.
+        let collection_generation_id = tables::utils::get_collection_generation_id(collection_spec)
+            .unwrap_or(models::Id::zero());
+
         let flow::CollectionSpec {
             ack_template_json: _,
             derivation: _,
@@ -143,7 +149,7 @@ impl Binding {
             read_schema_json: _,
             uuid_ptr,
             write_schema_json,
-        } = collection.as_ref().context("missing collection")?;
+        } = collection_spec;
 
         let document_uuid_ptr = doc::Pointer::from(uuid_ptr);
         let key_extractors = extractors::for_key(&key, &projections, &ser_policy)?;
@@ -152,6 +158,7 @@ impl Binding {
 
         Ok(Self {
             collection_name: name.clone(),
+            collection_generation_id,
             document_uuid_ptr,
             key_extractors,
             partition_extractors,

crates/runtime/src/capture/task.rs

@@ -19,6 +19,8 @@ impl<T: Stream<Item = anyhow::Result<Response>> + Send + 'static> ResponseStream
 pub struct Task {
     // Target collection.
     collection_name: String,
+    /// The generation id of the derived collection, which gets output as part of inferred schema updates.
+    collection_generation_id: models::Id,
     // JSON pointer at which document UUIDs are added.
     document_uuid_ptr: doc::Pointer,
     // Key components which are extracted from written documents.

crates/runtime/src/derive/mod.rs

@@ -365,6 +365,7 @@ pub async fn recv_connector_started_commit(
         tracing::info!(
             schema = ?::ops::DebugJson(serialized),
             collection_name = %task.collection_name,
+            collection_generation_id = %task.collection_generation_id,
             "inferred schema updated"
         );
     }

crates/runtime/src/derive/protocol.rs

@@ -14,6 +14,10 @@ impl Task {
 
         let response::Opened {} = opened.opened.as_ref().context("expected Opened")?;
 
+        let collection_spec = collection.context("missing collection")?;
+        let collection_generation_id =
+            tables::utils::get_collection_generation_id(&collection_spec)
+                .unwrap_or(models::Id::zero());
         let flow::CollectionSpec {
             ack_template_json: _,
             derivation,
@@ -25,7 +29,7 @@ impl Task {
             read_schema_json: _,
             uuid_ptr,
             write_schema_json,
-        } = collection.context("missing collection")?;
+        } = collection_spec;
 
         let flow::collection_spec::Derivation {
             config_json: _,
@@ -66,6 +70,7 @@ impl Task {
 
         Ok(Self {
             collection_name,
+            collection_generation_id,
             document_uuid_ptr,
             key_extractors,
             partition_extractors,

crates/runtime/src/derive/task.rs

@@ -1,7 +1,20 @@
 use models::{SourceCapture, SourceCaptureSchemaMode};
+use proto_flow::flow;
 use serde::{Deserialize, Serialize};
 use serde_json::Value;
 
+/// Returns the generation id for a given collection spec, if specified. Legacy
+/// collection specs may not have a generation id, in which case `None` is
+/// returned.
+pub fn get_collection_generation_id(c: &flow::CollectionSpec) -> Option<models::Id> {
+    let partition_prefix = &c.partition_template.as_ref()?.name;
+    let (_, last) = partition_prefix.rsplit_once('/')?;
+
+    // If this is a legacy collection spec, then the `last` path component will
+    // be a string that cannot be parsed as an id.
+    models::Id::from_hex(last).ok()
+}
+
 #[derive(Serialize, Deserialize)]
 pub struct ResourceSpecPointers {
     pub x_collection_name: doc::Pointer,

crates/tables/src/utils.rs

@@ -46,7 +46,7 @@
       "schema": {"$defs":{"__flowInline1":{"$id":"file:///Users/phil/projects/flow/ops-catalog/logs.schema.yaml","$schema":"https://json-schema.org/draft-07/schema","description":"Logs related to the processing of a Flow capture, derivation, or materialization","properties":{"fields":{"additionalProperties":true,"description":"Map of keys and values that are associated with this log entry.","type":"object"},"level":{"enum":["error","warn","info","debug","trace"]},"message":{"type":"string"},"shard":{"$ref":"shard.schema.yaml"},"ts":{"description":"Timestamp corresponding to the start of the transaction","format":"date-time","type":"string"}},"required":["shard","ts","level"],"title":"Flow task logs","type":"object"},"__flowInline2":{"$id":"file:///Users/phil/projects/flow/ops-catalog/shard.schema.yaml","$schema":"https://json-schema.org/draft-07/schema","description":"Identifies a specific shard of a task, which may be the source of a log message or metrics","properties":{"build":{"description":"The id of the build that this shard was running at the time the log was written","pattern":"[0-9a-f]{16}","type":"string"},"keyBegin":{"description":"The inclusive beginning of the shard's assigned key range","pattern":"[0-9a-f]{8}","type":"string"},"kind":{"description":"The type of the catalog task","enum":["capture","derivation","materialization"]},"name":{"description":"The name of the catalog task (without the task type prefix)","type":"string"},"rClockBegin":{"description":"The inclusive beginning of the shard's assigned rClock range","pattern":"[0-9a-f]{8}","type":"string"}},"required":["kind","name","keyBegin","rClockBegin"],"title":"Flow shard id","type":"object"}},"$id":"file:///Users/phil/projects/flow/ops-catalog/events.schema.yaml","$ref":"logs.schema.yaml","$schema":"https://json-schema.org/draft-07/schema","description":"Events are special logs that are intended to be consumed by the control plane","properties":{"fields":{"additionalProperties":true,"properties":{"error":{"description":"If the event represents an error, this field contains the error message.\n","type":"string"},"eventTarget":{"description":"The target of the event is a catalog name that the event pertains to.\n","type":"string"},"eventType":{"description":"Identifies this log message as an event of the given type. Events\nare special logs that are meant to be observed by the Flow control plane.\n","type":"string"}},"required":["eventType","eventTarget"]},"shard":{"description":"The source of the event, which may differ from the eventTarget"}},"required":["fields"],"title":"Flow events"},
       "key": [
         "/fields/eventTarget",
-        "/eventType"
+        "/fields/eventType"
       ],
       "projections": {
         "event_type": {
@@ -105,7 +105,7 @@
                 "/shard/name"
               ]
             },
-            "lambda": "select\n  $fields->>'collection_name' as collection_name,\n  $fields->'schema' as schema\nwhere $message = 'inferred schema updated';\n"
+            "lambda": "select\n  $fields->>'collection_name' as collection_name,\n  coalesce($fields->>'collection_generation_id', '0000000000000000') as collection_generation_id,\n  $fields->'schema' as schema\nwhere $message = 'inferred schema updated';\n"
           }
         ],
         "shards": {