Feature/UUID timestamp extraction

Cargo.toml

@@ -176,6 +176,7 @@ yaml-merge-keys = { version = "0.5", features = ["serde_yaml"] }
 zip = "0.5"
 zstd = "0.11.2"
 derivative = "2.2.0"
+const_format = "0.2.30"
 
 # Used exclusively as dev-dependencies
 assert_cmd = "2.0"

crates/assemble/src/lib.rs

@@ -9,6 +9,9 @@ use std::time::Duration;
 mod ops;
 pub use ops::generate_ops_collections;
 
+// For the forseeable future, we don't allow customizing this.
+pub const UUID_PTR: &str = "/_meta/uuid";
+
 pub fn inference(shape: &Shape, exists: Exists) -> flow::Inference {
     let default_json = shape
         .default
@@ -422,9 +425,6 @@ pub fn collection_spec(
         })
         .collect();
 
-    // For the forseeable future, we don't allow customizing this.
-    let uuid_ptr = "/_meta/uuid".to_string();
-
     let (write_schema_json, read_schema_json) = match (schema, write_schema, read_schema) {
         (Some(schema), None, None) => (schema.to_string(), String::new()),
         (None, Some(write_schema), Some(read_schema)) => {
@@ -440,7 +440,7 @@ pub fn collection_spec(
         key: key.iter().map(|p| p.to_string()).collect(),
         projections,
         partition_fields,
-        uuid_ptr,
+        uuid_ptr: UUID_PTR.to_string(),
         ack_template_json: serde_json::json!({
                 "_meta": {"uuid": "DocUUIDPlaceholder-329Bb50aa48EAa9ef",
                 "ack": true,

crates/derive-sqlite/Cargo.toml

@@ -11,6 +11,7 @@ license.workspace = true
 [dependencies]
 doc = { path = "../doc" }
 json = { path = "../json" }
+derive = { path = "../derive" }
 proto-flow = { path = "../proto-flow" }
 
 anyhow = { workspace = true }

crates/derive-sqlite/src/connector.rs

@@ -1,3 +1,5 @@
+use std::iter;
+
 use super::{dbutil, do_validate, parse_validate, Config, Lambda, Param, Transform};
 use anyhow::Context;
 use futures::channel::mpsc;
@@ -213,7 +215,12 @@ fn parse_open(
             } = transform;
 
             let source = source.unwrap();
-            let params: Vec<_> = source.projections.iter().map(Param::new).collect();
+            let params: Vec<_> = source
+                .projections
+                .iter()
+                .zip(iter::repeat(&source))
+                .map(|(p, c)| Param::new(p, c))
+                .collect();
 
             let block: String = serde_json::from_str(&lambda_config_json).with_context(|| {
                 format!("failed to parse SQLite lambda block: {lambda_config_json}")

crates/derive-sqlite/src/lambda.rs

@@ -141,13 +141,16 @@ fn bind_parameter<N: doc::AsNode>(
         is_content_encoding_base64,
         is_format_integer,
         is_format_number,
+        collection,
         ..
     }: &Param,
     document: &N,
 ) -> rusqlite::Result<()> {
+    use derive::PointerExt;
     use doc::Node;
+    let uuid_ptr_parsed = doc::Pointer::maybe_parse(collection.uuid_ptr.as_str());
 
-    match ptr.query(document).map(doc::AsNode::as_node) {
+    ptr.query_and_resolve_virtuals(uuid_ptr_parsed, document, |doc| match doc {
         None | Some(Node::Null) => return stmt.raw_bind_parameter(index + 1, None::<bool>),
         Some(Node::Bool(b)) => return stmt.raw_bind_parameter(index + 1, b),
 
@@ -179,7 +182,7 @@ fn bind_parameter<N: doc::AsNode>(
         Some(n @ Node::Object(_)) => {
             stmt.raw_bind_parameter(index + 1, &serde_json::to_string(&n).unwrap())
         }
-    }
+    })
 }
 
 fn row_to_json(

crates/derive-sqlite/src/lib.rs

@@ -68,24 +68,27 @@ fn test_param(
 ) -> Param {
     use proto_flow::flow;
 
-    Param::new(&flow::Projection {
-        field: field.to_string(),
-        ptr: ptr.to_string(),
-        inference: Some(flow::Inference {
-            string: Some(flow::inference::String {
-                format: if is_format_integer {
-                    "integer"
-                } else if is_format_number {
-                    "number"
-                } else {
-                    ""
-                }
-                .to_string(),
-                content_encoding: if is_base64 { "base64" } else { "" }.to_string(),
+    Param::new(
+        &flow::Projection {
+            field: field.to_string(),
+            ptr: ptr.to_string(),
+            inference: Some(flow::Inference {
+                string: Some(flow::inference::String {
+                    format: if is_format_integer {
+                        "integer"
+                    } else if is_format_number {
+                        "number"
+                    } else {
+                        ""
+                    }
+                    .to_string(),
+                    content_encoding: if is_base64 { "base64" } else { "" }.to_string(),
+                    ..Default::default()
+                }),
                 ..Default::default()
             }),
             ..Default::default()
-        }),
-        ..Default::default()
-    })
+        },
+        &flow::CollectionSpec::default(),
+    )
 }

crates/derive-sqlite/src/param.rs

@@ -7,6 +7,8 @@ use proto_flow::flow;
 pub struct Param {
     // Projection which is being related as a SQLite parameter.
     pub projection: flow::Projection,
+    // Collection which contains the projection that is being related as a SQLite parameter.
+    pub collection: flow::CollectionSpec,
     // Canonical SQLite parameter encoding for this field.
     pub canonical_encoding: String,
     // JSON pointer location of this field within documents.
@@ -20,9 +22,10 @@ pub struct Param {
 }
 
 impl Param {
-    pub fn new(p: &flow::Projection) -> Self {
+    pub fn new(p: &flow::Projection, c: &flow::CollectionSpec) -> Self {
         Self {
             projection: p.clone(),
+            collection: c.clone(),
             canonical_encoding: canonical_param_encoding(&p.field),
             ptr: doc::Pointer::from_str(&p.ptr),
             is_format_integer: matches!(&p.inference, Some(flow::Inference{string: Some(str), ..}) if str.format == "integer"),

crates/derive-sqlite/src/validate.rs

@@ -1,3 +1,5 @@
+use std::iter;
+
 use super::{dbutil, is_url_to_generate, Config, Param, Transform};
 use anyhow::Context;
 use proto_flow::{
@@ -33,7 +35,12 @@ pub fn parse_validate(
             } = transform;
 
             let source = source.unwrap();
-            let params: Vec<_> = source.projections.iter().map(Param::new).collect();
+            let params: Vec<_> = source
+                .projections
+                .iter()
+                .zip(iter::repeat(&source))
+                .map(|(p, c)| Param::new(p, c))
+                .collect();
 
             let block: String = serde_json::from_str(&lambda_config_json).with_context(|| {
                 format!("failed to parse SQLite lambda block: {lambda_config_json}")

crates/derive/Cargo.toml

@@ -33,6 +33,7 @@ thiserror = { workspace = true }
 tracing = { workspace = true }
 url = { workspace = true }
 uuid = { workspace = true }
+time = { workspace = true }
 
 [dev-dependencies]
 insta = { workspace = true }