chore: upgrade sqltk to 0.11.0

Cargo.toml

@@ -42,7 +42,7 @@ strip = "none"
 debug = true
 
 [workspace.dependencies]
-sqltk = { version = "0.10.0" }
+sqltk = { version = "0.11.0" }
 cipherstash-client = { version = "=0.42.0" }
 cipherstash-config = { version = "=0.42.0" }
 cts-common = { version = "=0.42.0" }

mise.toml

@@ -1,10 +1,10 @@
 [settings]
-# Config for test environments
-# Can be invoked with: mise --env tcp run <task>
+# Config for test environments. Anchor these paths to this config so they are
+# trusted regardless of the task's working directory.
 trusted_config_paths = [
-  "./tests/mise.toml",
-  "./tests/mise.tcp.toml",
-  "./tests/mise.tls.toml",
+  "{{config_root}}/tests/mise.toml",
+  "{{config_root}}/tests/mise.tcp.toml",
+  "{{config_root}}/tests/mise.tls.toml",
 ]
 
 [task_config]

packages/cipherstash-proxy-integration/src/insert/insert_on_conflict.rs

@@ -0,0 +1,25 @@
+#[cfg(test)]
+mod tests {
+    use crate::common::{assert_encrypted_text, clear, execute_query, query_by, random_id, trace};
+
+    #[tokio::test]
+    async fn conflict_update_encrypts_excluded_value() {
+        trace();
+        clear().await;
+
+        let id = random_id();
+        let initial = "initial value".to_string();
+        let updated = "updated value".to_string();
+        let sql = "INSERT INTO encrypted (id, encrypted_text) VALUES ($1, $2) \
+                   ON CONFLICT (id) DO UPDATE SET encrypted_text = excluded.encrypted_text";
+
+        execute_query(sql, &[&id, &initial]).await;
+        execute_query(sql, &[&id, &updated]).await;
+
+        assert_eq!(
+            query_by::<String>("SELECT encrypted_text FROM encrypted WHERE id = $1", &id).await,
+            vec![updated.clone()]
+        );
+        assert_encrypted_text(id, "encrypted_text", &updated).await;
+    }
+}

packages/cipherstash-proxy-integration/src/insert/insert_with_params.rs

@@ -1,9 +1,8 @@
 #[cfg(test)]
 mod tests {
-    use crate::common::{clear, insert, query, random_id, random_limited, trace};
+    use crate::common::{clear, insert, random_id, random_limited, trace};
     use chrono::NaiveDate;
     use rand::{seq::IndexedRandom, Rng};
-    use serde_json::Value;
     use tokio_postgres::types::ToSql;
     use tracing::info;
 
@@ -38,8 +37,8 @@ mod tests {
     /// Return as a tuple of two vecs:
     ///     - first vec contains column names
     ///     - second vec contains values of the corresponding column type
-    pub fn generate_columns_with_values() -> (Vec<String>, Vec<Box<(dyn ToSql + Sync)>>) {
-        let columns = vec![
+    pub fn generate_columns_with_values() -> (Vec<String>, Vec<Box<dyn ToSql + Sync>>) {
+        let columns = [
             ("i16", "int2"),
             ("i32", "int4"),
             ("i64", "int8"),
@@ -68,14 +67,6 @@ mod tests {
         (columns, values)
     }
 
-    pub async fn query<T: for<'a> tokio_postgres::types::FromSql<'a> + Send + Sync>(
-        sql: &str,
-    ) -> Vec<T> {
-        let client = connect_with_tls(*PROXY).await;
-        let rows = client.query(sql, &[]).await.unwrap();
-        rows.iter().map(|row| row.get(0)).collect::<Vec<T>>()
-    }
-
     #[tokio::test]
     pub async fn test_everything_all_at_once() {
         trace();
@@ -99,12 +90,6 @@ mod tests {
 
         info!(sql);
         insert(&sql, &params).await;
-
-        let sql = format!("SELECT {columns} FROM encrypted WHERE id = $1");
-
-        // let actual = query_by::<$type>(&sql, &id).await;
-
-        // assert_eq!(expected, actual);
     }
 
     // test_insert_with_params!(insert_with_params_int2, i16, int2);

packages/cipherstash-proxy-integration/src/insert/mod.rs

@@ -1,5 +1,7 @@
 mod insert_domain_type;
+mod insert_on_conflict;
 mod insert_with_literal;
 mod insert_with_null_literal;
 mod insert_with_null_param;
 mod insert_with_param;
+mod insert_with_params;

packages/eql-mapper/src/function_arg.rs

@@ -0,0 +1,28 @@
+use sqltk::parser::ast::{Expr, FunctionArg, FunctionArgExpr};
+
+pub(crate) fn function_arg_expr(arg: &FunctionArg) -> &FunctionArgExpr {
+    match arg {
+        FunctionArg::Named { arg, .. } => arg,
+        FunctionArg::ExprNamed { arg, .. } => arg,
+        FunctionArg::Unnamed(arg) => arg,
+    }
+}
+
+pub(crate) fn function_arg_value(arg: &FunctionArg) -> Option<&Expr> {
+    match function_arg_expr(arg) {
+        FunctionArgExpr::Expr(expr) => Some(expr),
+        FunctionArgExpr::QualifiedWildcard(_) | FunctionArgExpr::Wildcard => None,
+    }
+}
+
+pub(crate) fn function_arg_value_mut(arg: &mut FunctionArg) -> Option<&mut Expr> {
+    let arg = match arg {
+        FunctionArg::Named { arg, .. } => arg,
+        FunctionArg::ExprNamed { arg, .. } => arg,
+        FunctionArg::Unnamed(arg) => arg,
+    };
+    match arg {
+        FunctionArgExpr::Expr(expr) => Some(expr),
+        FunctionArgExpr::QualifiedWildcard(_) | FunctionArgExpr::Wildcard => None,
+    }
+}

packages/eql-mapper/src/importer.rs

@@ -5,8 +5,8 @@ use crate::{
     Relation, ScopeError, ScopeTracker,
 };
 use sqltk::parser::ast::{
-    Cte, Ident, Insert, ObjectNamePart, OnConflict, OnConflictAction, OnInsert, TableAlias,
-    TableFactor, TableObject,
+    Cte, Ident, Insert, ObjectNamePart, OnConflict, OnConflictAction, TableAlias, TableFactor,
+    TableObject,
 };
 use sqltk::{Break, Visitable, Visitor};
 use std::{cell::RefCell, fmt::Debug, marker::PhantomData, ops::ControlFlow, rc::Rc, sync::Arc};
@@ -18,6 +18,8 @@ pub struct Importer<'ast> {
     table_resolver: Arc<TableResolver>,
     registry: Rc<RefCell<TypeRegistry<'ast>>>,
     scope_tracker: Rc<RefCell<ScopeTracker<'ast>>>,
+    insert_projections: Vec<Arc<Type>>,
+    shadowed_excluded_relations: Vec<Option<Rc<Relation>>>,
     _ast: PhantomData<&'ast ()>,
 }
 
@@ -31,21 +33,27 @@ impl<'ast> Importer<'ast> {
             registry: registry.into(),
             table_resolver: table_resolver.into(),
             scope_tracker: scope.into(),
+            insert_projections: Vec::new(),
+            shadowed_excluded_relations: Vec::new(),
             _ast: PhantomData,
         }
     }
 
-    fn update_scope_for_insert_statement(&mut self, insert: &Insert) -> Result<(), ImportError> {
+    fn update_scope_for_insert_statement(
+        &mut self,
+        insert: &Insert,
+    ) -> Result<Arc<Type>, ImportError> {
         if let Insert {
             table: TableObject::TableName(table_name),
             table_alias,
-            on,
             ..
         } = insert
         {
             let table = self.table_resolver.resolve_table(table_name)?;
 
-            let projection = Projection::new_from_schema_table(table.clone())?;
+            let projection = Arc::new(Type::Value(Value::Projection(
+                Projection::new_from_schema_table(table.clone())?,
+            )));
 
             // The relation is named — by its alias when one is written, by the
             // table name otherwise — so that qualified references (`t.col` in
@@ -57,30 +65,10 @@ impl<'ast> Importer<'ast> {
 
             self.scope_tracker.borrow_mut().add_relation(Relation {
                 name,
-                projection_type: Type::Value(Value::Projection(projection.clone())).into(),
+                projection_type: Arc::clone(&projection),
             })?;
 
-            // `ON CONFLICT DO UPDATE` can read the row proposed for insertion
-            // through the `excluded` pseudo-table, which projects exactly the
-            // target table's columns. Bringing it into scope is what gives
-            // `excluded.<col>` a type — including the column's EQL type, so an
-            // upsert like `SET enc = excluded.enc` is fully constrained.
-            //
-            // An unqualified column reference in the `DO UPDATE` expressions is
-            // now ambiguous (both relations project it), which mirrors
-            // PostgreSQL's own `column reference is ambiguous` error there.
-            if let Some(OnInsert::OnConflict(OnConflict {
-                action: OnConflictAction::DoUpdate(_),
-                ..
-            })) = on
-            {
-                self.scope_tracker.borrow_mut().add_relation(Relation {
-                    name: Some(Ident::new("excluded")),
-                    projection_type: Type::Value(Value::Projection(projection)).into(),
-                })?;
-            }
-
-            Ok(())
+            Ok(projection)
         } else {
             Err(ImportError::Unsupported(
                 "unsupported TableObject variant in Insert".to_string(),
@@ -322,8 +310,8 @@ pub enum ImportError {
     #[error(transparent)]
     ScopeError(#[from] ScopeError),
 
-    #[error("Expected projection")]
-    ExpectedProjection,
+    #[error("Importer traversal invariant failed: {0}")]
+    TraversalInvariant(&'static str),
 
     #[error(transparent)]
     TypeError(#[from] TypeError),
@@ -342,15 +330,68 @@ impl<'ast> Visitor<'ast> for Importer<'ast> {
         // 2. Child nodes of the `Insert` need to resolve identifiers in the context of the scope, so exit would be too
         // late.
         if let Some(insert) = node.downcast_ref::<Insert>() {
-            if let Err(err) = self.update_scope_for_insert_statement(insert) {
-                return ControlFlow::Break(Break::Err(err));
+            match self.update_scope_for_insert_statement(insert) {
+                Ok(projection) => self.insert_projections.push(projection),
+                Err(err) => return ControlFlow::Break(Break::Err(err)),
+            }
+        }
+
+        // `excluded` exists only inside `ON CONFLICT DO UPDATE`. Adding it at
+        // the clause boundary keeps it visible to assignments and the WHERE
+        // predicate, but not to the INSERT source or RETURNING clause.
+        if let Some(on_conflict) = node.downcast_ref::<OnConflict>() {
+            if on_conflict_is_update(on_conflict) {
+                let Some(projection_type) = self.insert_projections.last().cloned() else {
+                    return ControlFlow::Break(Break::Err(ImportError::TraversalInvariant(
+                        "ON CONFLICT DO UPDATE has no enclosing INSERT projection",
+                    )));
+                };
+
+                match self
+                    .scope_tracker
+                    .borrow_mut()
+                    .add_shadowing_relation(Relation {
+                        name: Some(Ident::new("excluded")),
+                        projection_type,
+                    }) {
+                    Ok(shadowed) => self.shadowed_excluded_relations.push(shadowed),
+                    Err(err) => return ControlFlow::Break(Break::Err(err.into())),
+                }
             }
         }
 
         ControlFlow::Continue(())
     }
 
     fn exit<N: Visitable>(&mut self, node: &'ast N) -> ControlFlow<Break<Self::Error>> {
+        if let Some(on_conflict) = node.downcast_ref::<OnConflict>() {
+            if on_conflict_is_update(on_conflict) {
+                // Remove the pseudo-relation added on entry before traversal
+                // continues into the INSERT's RETURNING clause, restoring a
+                // target table binding that it temporarily shadowed.
+                let Some(shadowed) = self.shadowed_excluded_relations.pop() else {
+                    return ControlFlow::Break(Break::Err(ImportError::TraversalInvariant(
+                        "ON CONFLICT DO UPDATE exited without a shadow record",
+                    )));
+                };
+                if let Err(err) = self
+                    .scope_tracker
+                    .borrow_mut()
+                    .remove_shadowing_relation(&Ident::new("excluded"), shadowed)
+                {
+                    return ControlFlow::Break(Break::Err(err.into()));
+                }
+            }
+        }
+
+        if let Some(_insert) = node.downcast_ref::<Insert>() {
+            if self.insert_projections.pop().is_none() {
+                return ControlFlow::Break(Break::Err(ImportError::TraversalInvariant(
+                    "INSERT exited without a matching projection",
+                )));
+            }
+        }
+
         if let Some(cte) = node.downcast_ref::<Cte>() {
             if let Err(err) = self.update_scope_for_cte(cte) {
                 return ControlFlow::Break(Break::Err(err));
@@ -366,3 +407,7 @@ impl<'ast> Visitor<'ast> for Importer<'ast> {
         ControlFlow::Continue(())
     }
 }
+
+fn on_conflict_is_update(on_conflict: &OnConflict) -> bool {
+    matches!(on_conflict.action, OnConflictAction::DoUpdate(_))
+}