diff --git a/Cargo.lock b/Cargo.lock
index 13f82e6f..2be24067 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1749,6 +1749,7 @@ dependencies = [
  "mongodb",
  "mongodb-agent-common",
  "mongodb-support",
+ "ndc-models",
  "proptest",
  "serde",
  "serde_json",
@@ -1817,14 +1818,16 @@ dependencies = [
 
 [[package]]
 name = "ndc-models"
-version = "0.1.4"
-source = "git+http://github.com/hasura/ndc-spec.git?tag=v0.1.4#20172e3b2552b78d16dbafcd047f559ced420309"
+version = "0.1.5"
+source = "git+http://github.com/hasura/ndc-spec.git?tag=v0.1.5#78f52768bd02a8289194078a5abc2432c8e3a758"
 dependencies = [
  "indexmap 2.2.6",
+ "ref-cast",
  "schemars",
  "serde",
  "serde_json",
  "serde_with 3.8.1",
+ "smol_str",
 ]
 
 [[package]]
@@ -1841,14 +1844,15 @@ dependencies = [
  "ndc-test-helpers",
  "nonempty",
  "pretty_assertions",
+ "ref-cast",
  "serde_json",
  "thiserror",
 ]
 
 [[package]]
 name = "ndc-sdk"
-version = "0.1.4"
-source = "git+https://github.com/hasura/ndc-sdk-rs.git?tag=v0.1.4#29adcb5983c1237e8a5f4732d5230c2ba8ab75d3"
+version = "0.2.1"
+source = "git+https://github.com/hasura/ndc-sdk-rs.git?tag=v0.2.1#83a906e8a744ee78d84aeee95f61bf3298a982ea"
 dependencies = [
  "async-trait",
  "axum",
@@ -1880,8 +1884,8 @@ dependencies = [
 
 [[package]]
 name = "ndc-test"
-version = "0.1.4"
-source = "git+http://github.com/hasura/ndc-spec.git?tag=v0.1.4#20172e3b2552b78d16dbafcd047f559ced420309"
+version = "0.1.5"
+source = "git+http://github.com/hasura/ndc-spec.git?tag=v0.1.5#78f52768bd02a8289194078a5abc2432c8e3a758"
 dependencies = [
  "async-trait",
  "clap",
@@ -1893,6 +1897,7 @@ dependencies = [
  "semver 1.0.23",
  "serde",
  "serde_json",
+ "smol_str",
  "thiserror",
  "tokio",
  "url",
@@ -2400,6 +2405,26 @@ dependencies = [
  "bitflags 2.5.0",
 ]
 
+[[package]]
+name = "ref-cast"
+version = "1.0.23"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ccf0a6f84d5f1d581da8b41b47ec8600871962f2a528115b542b362d4b744931"
+dependencies = [
+ "ref-cast-impl",
+]
+
+[[package]]
+name = "ref-cast-impl"
+version = "1.0.23"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bcc303e793d3734489387d205e9b186fac9c6cfacedd98cbb2e8a5943595f3e6"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.66",
+]
+
 [[package]]
 name = "regex"
 version = "1.10.5"
@@ -4163,9 +4188,9 @@ checksum = "ced3678a2879b30306d323f4542626697a464a97c0a07c9aebf7ebca65cd4dde"
 
 [[package]]
 name = "zerovec"
-version = "0.10.2"
+version = "0.10.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bb2cc8827d6c0994478a15c53f374f46fbd41bea663d809b14744bc42e6b109c"
+checksum = "aa2b893d79df23bfb12d5461018d408ea19dfafe76c2c7ef6d4eba614f8ff079"
 dependencies = [
  "yoke",
  "zerofrom",
@@ -4174,9 +4199,9 @@ dependencies = [
 
 [[package]]
 name = "zerovec-derive"
-version = "0.10.2"
+version = "0.10.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "97cf56601ee5052b4417d90c8755c6683473c926039908196cf35d99f893ebe7"
+checksum = "6eafa6dfb17584ea3e2bd6e76e0cc15ad7af12b09abdd1ca55961bed9b1063c6"
 dependencies = [
  "proc-macro2",
  "quote",
diff --git a/Cargo.toml b/Cargo.toml
index 765d715b..a59eb2e9 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -18,8 +18,8 @@ resolver = "2"
 # The tag or rev of ndc-models must match the locked tag or rev of the
 # ndc-models dependency of ndc-sdk
 [workspace.dependencies]
-ndc-sdk = { git = "https://github.com/hasura/ndc-sdk-rs.git", tag = "v0.1.4" }
-ndc-models = { git = "http://github.com/hasura/ndc-spec.git", tag = "v0.1.4" }
+ndc-sdk = { git = "https://github.com/hasura/ndc-sdk-rs.git", tag = "v0.2.1" }
+ndc-models = { git = "http://github.com/hasura/ndc-spec.git", tag = "v0.1.5" }
 
 indexmap = { version = "2", features = [
   "serde",
@@ -27,6 +27,7 @@ indexmap = { version = "2", features = [
 itertools = "^0.12.1"
 mongodb = { version = "2.8", features = ["tracing-unstable"] }
 schemars = "^0.8.12"
+ref-cast = "1.0.23"
 
 # Connecting to MongoDB Atlas database with time series collections fails in the
 # latest released version of the MongoDB Rust driver. A fix has been merged, but
diff --git a/crates/cli/Cargo.toml b/crates/cli/Cargo.toml
index fb59274f..031d7891 100644
--- a/crates/cli/Cargo.toml
+++ b/crates/cli/Cargo.toml
@@ -14,6 +14,7 @@ clap = { version = "4.5.1", features = ["derive", "env"] }
 futures-util = "0.3.28"
 indexmap = { workspace = true }
 itertools = { workspace = true }
+ndc-models = { workspace = true }
 serde = { version = "1.0", features = ["derive"] }
 serde_json = { version = "1.0.113", features = ["raw_value"] }
 thiserror = "1.0.57"
diff --git a/crates/cli/src/introspection/sampling.rs b/crates/cli/src/introspection/sampling.rs
index 51a5f720..c01360ca 100644
--- a/crates/cli/src/introspection/sampling.rs
+++ b/crates/cli/src/introspection/sampling.rs
@@ -12,8 +12,8 @@ use mongodb::bson::{doc, Bson, Document};
 use mongodb_agent_common::state::ConnectorState;
 use mongodb_support::BsonScalarType::{self, *};
 
-type ObjectField = WithName<schema::ObjectField>;
-type ObjectType = WithName<schema::ObjectType>;
+type ObjectField = WithName<ndc_models::FieldName, schema::ObjectField>;
+type ObjectType = WithName<ndc_models::ObjectTypeName, schema::ObjectType>;
 
 /// Sample from all collections in the database and return a Schema.
 /// Return an error if there are any errors accessing the database
@@ -66,7 +66,7 @@ async fn sample_schema_from_collection(
     let is_collection_type = true;
     while let Some(document) = cursor.try_next().await? {
         let object_types = make_object_type(
-            collection_name,
+            &collection_name.into(),
             &document,
             is_collection_type,
             all_schema_nullable,
@@ -81,10 +81,10 @@ async fn sample_schema_from_collection(
         Ok(None)
     } else {
         let collection_info = WithName::named(
-            collection_name.to_string(),
+            collection_name.into(),
             schema::Collection {
                 description: None,
-                r#type: collection_name.to_string(),
+                r#type: collection_name.into(),
             },
         );
         Ok(Some(Schema {
@@ -95,7 +95,7 @@ async fn sample_schema_from_collection(
 }
 
 fn make_object_type(
-    object_type_name: &str,
+    object_type_name: &ndc_models::ObjectTypeName,
     document: &Document,
     is_collection_type: bool,
     all_schema_nullable: bool,
@@ -118,7 +118,7 @@ fn make_object_type(
     };
 
     let object_type = WithName::named(
-        object_type_name.to_string(),
+        object_type_name.to_owned(),
         schema::ObjectType {
             description: None,
             fields: WithName::into_map(object_fields),
@@ -140,7 +140,7 @@ fn make_object_field(
     let (collected_otds, field_type) =
         make_field_type(&object_type_name, field_value, all_schema_nullable);
     let object_field_value = WithName::named(
-        field_name.to_owned(),
+        field_name.into(),
         schema::ObjectField {
             description: None,
             r#type: field_type,
@@ -161,7 +161,10 @@ pub fn type_from_bson(
     object_type_name: &str,
     value: &Bson,
     all_schema_nullable: bool,
-) -> (BTreeMap<std::string::String, schema::ObjectType>, Type) {
+) -> (
+    BTreeMap<ndc_models::ObjectTypeName, schema::ObjectType>,
+    Type,
+) {
     let (object_types, t) = make_field_type(object_type_name, value, all_schema_nullable);
     (WithName::into_map(object_types), t)
 }
@@ -196,7 +199,7 @@ fn make_field_type(
         Bson::Document(document) => {
             let is_collection_type = false;
             let collected_otds = make_object_type(
-                object_type_name,
+                &object_type_name.into(),
                 document,
                 is_collection_type,
                 all_schema_nullable,
@@ -238,24 +241,28 @@ mod tests {
 
     #[test]
     fn simple_doc() -> Result<(), anyhow::Error> {
-        let object_name = "foo";
+        let object_name = "foo".into();
         let doc = doc! {"my_int": 1, "my_string": "two"};
-        let result =
-            WithName::into_map::<BTreeMap<_, _>>(make_object_type(object_name, &doc, false, false));
+        let result = WithName::into_map::<BTreeMap<_, _>>(make_object_type(
+            &object_name,
+            &doc,
+            false,
+            false,
+        ));
 
         let expected = BTreeMap::from([(
             object_name.to_owned(),
             ObjectType {
                 fields: BTreeMap::from([
                     (
-                        "my_int".to_owned(),
+                        "my_int".into(),
                         ObjectField {
                             r#type: Type::Scalar(BsonScalarType::Int),
                             description: None,
                         },
                     ),
                     (
-                        "my_string".to_owned(),
+                        "my_string".into(),
                         ObjectField {
                             r#type: Type::Scalar(BsonScalarType::String),
                             description: None,
@@ -273,31 +280,31 @@ mod tests {
 
     #[test]
     fn simple_doc_nullable_fields() -> Result<(), anyhow::Error> {
-        let object_name = "foo";
+        let object_name = "foo".into();
         let doc = doc! {"my_int": 1, "my_string": "two", "_id": 0};
         let result =
-            WithName::into_map::<BTreeMap<_, _>>(make_object_type(object_name, &doc, true, true));
+            WithName::into_map::<BTreeMap<_, _>>(make_object_type(&object_name, &doc, true, true));
 
         let expected = BTreeMap::from([(
             object_name.to_owned(),
             ObjectType {
                 fields: BTreeMap::from([
                     (
-                        "_id".to_owned(),
+                        "_id".into(),
                         ObjectField {
                             r#type: Type::Scalar(BsonScalarType::Int),
                             description: None,
                         },
                     ),
                     (
-                        "my_int".to_owned(),
+                        "my_int".into(),
                         ObjectField {
                             r#type: Type::Nullable(Box::new(Type::Scalar(BsonScalarType::Int))),
                             description: None,
                         },
                     ),
                     (
-                        "my_string".to_owned(),
+                        "my_string".into(),
                         ObjectField {
                             r#type: Type::Nullable(Box::new(Type::Scalar(BsonScalarType::String))),
                             description: None,
@@ -315,32 +322,36 @@ mod tests {
 
     #[test]
     fn array_of_objects() -> Result<(), anyhow::Error> {
-        let object_name = "foo";
+        let object_name = "foo".into();
         let doc = doc! {"my_array": [{"foo": 42, "bar": ""}, {"bar": "wut", "baz": 3.77}]};
-        let result =
-            WithName::into_map::<BTreeMap<_, _>>(make_object_type(object_name, &doc, false, false));
+        let result = WithName::into_map::<BTreeMap<_, _>>(make_object_type(
+            &object_name,
+            &doc,
+            false,
+            false,
+        ));
 
         let expected = BTreeMap::from([
             (
-                "foo_my_array".to_owned(),
+                "foo_my_array".into(),
                 ObjectType {
                     fields: BTreeMap::from([
                         (
-                            "foo".to_owned(),
+                            "foo".into(),
                             ObjectField {
                                 r#type: Type::Nullable(Box::new(Type::Scalar(BsonScalarType::Int))),
                                 description: None,
                             },
                         ),
                         (
-                            "bar".to_owned(),
+                            "bar".into(),
                             ObjectField {
                                 r#type: Type::Scalar(BsonScalarType::String),
                                 description: None,
                             },
                         ),
                         (
-                            "baz".to_owned(),
+                            "baz".into(),
                             ObjectField {
                                 r#type: Type::Nullable(Box::new(Type::Scalar(
                                     BsonScalarType::Double,
@@ -356,7 +367,7 @@ mod tests {
                 object_name.to_owned(),
                 ObjectType {
                     fields: BTreeMap::from([(
-                        "my_array".to_owned(),
+                        "my_array".into(),
                         ObjectField {
                             r#type: Type::ArrayOf(Box::new(Type::Object(
                                 "foo_my_array".to_owned(),
@@ -376,32 +387,36 @@ mod tests {
 
     #[test]
     fn non_unifiable_array_of_objects() -> Result<(), anyhow::Error> {
-        let object_name = "foo";
+        let object_name = "foo".into();
         let doc = doc! {"my_array": [{"foo": 42, "bar": ""}, {"bar": 17, "baz": 3.77}]};
-        let result =
-            WithName::into_map::<BTreeMap<_, _>>(make_object_type(object_name, &doc, false, false));
+        let result = WithName::into_map::<BTreeMap<_, _>>(make_object_type(
+            &object_name,
+            &doc,
+            false,
+            false,
+        ));
 
         let expected = BTreeMap::from([
             (
-                "foo_my_array".to_owned(),
+                "foo_my_array".into(),
                 ObjectType {
                     fields: BTreeMap::from([
                         (
-                            "foo".to_owned(),
+                            "foo".into(),
                             ObjectField {
                                 r#type: Type::Nullable(Box::new(Type::Scalar(BsonScalarType::Int))),
                                 description: None,
                             },
                         ),
                         (
-                            "bar".to_owned(),
+                            "bar".into(),
                             ObjectField {
                                 r#type: Type::ExtendedJSON,
                                 description: None,
                             },
                         ),
                         (
-                            "baz".to_owned(),
+                            "baz".into(),
                             ObjectField {
                                 r#type: Type::Nullable(Box::new(Type::Scalar(
                                     BsonScalarType::Double,
@@ -417,7 +432,7 @@ mod tests {
                 object_name.to_owned(),
                 ObjectType {
                     fields: BTreeMap::from([(
-                        "my_array".to_owned(),
+                        "my_array".into(),
                         ObjectField {
                             r#type: Type::ArrayOf(Box::new(Type::Object(
                                 "foo_my_array".to_owned(),
diff --git a/crates/cli/src/introspection/type_unification.rs b/crates/cli/src/introspection/type_unification.rs
index bf997c3f..dd813f3c 100644
--- a/crates/cli/src/introspection/type_unification.rs
+++ b/crates/cli/src/introspection/type_unification.rs
@@ -12,10 +12,9 @@ use mongodb_support::{
     align::align,
     BsonScalarType::{self, *},
 };
-use std::string::String;
 
-type ObjectField = WithName<schema::ObjectField>;
-type ObjectType = WithName<schema::ObjectType>;
+type ObjectField = WithName<ndc_models::FieldName, schema::ObjectField>;
+type ObjectType = WithName<ndc_models::ObjectTypeName, schema::ObjectType>;
 
 /// Unify two types.
 /// This is computing the join (or least upper bound) of the two types in a lattice
@@ -94,14 +93,14 @@ pub fn make_nullable_field(field: ObjectField) -> ObjectField {
 /// Unify two `ObjectType`s.
 /// Any field that appears in only one of the `ObjectType`s will be made nullable.
 fn unify_object_type(object_type_a: ObjectType, object_type_b: ObjectType) -> ObjectType {
-    let field_map_a: IndexMap<String, ObjectField> = object_type_a
+    let field_map_a: IndexMap<ndc_models::FieldName, ObjectField> = object_type_a
         .value
         .fields
         .into_iter()
         .map_into::<ObjectField>()
         .map(|o| (o.name.to_owned(), o))
         .collect();
-    let field_map_b: IndexMap<String, ObjectField> = object_type_b
+    let field_map_b: IndexMap<ndc_models::FieldName, ObjectField> = object_type_b
         .value
         .fields
         .into_iter()
@@ -154,11 +153,11 @@ pub fn unify_object_types(
     object_types_a: Vec<ObjectType>,
     object_types_b: Vec<ObjectType>,
 ) -> Vec<ObjectType> {
-    let type_map_a: IndexMap<String, ObjectType> = object_types_a
+    let type_map_a: IndexMap<ndc_models::ObjectTypeName, ObjectType> = object_types_a
         .into_iter()
         .map(|t| (t.name.to_owned(), t))
         .collect();
-    let type_map_b: IndexMap<String, ObjectType> = object_types_b
+    let type_map_b: IndexMap<ndc_models::ObjectTypeName, ObjectType> = object_types_b
         .into_iter()
         .map(|t| (t.name.to_owned(), t))
         .collect();
@@ -303,26 +302,26 @@ mod tests {
             }
 
             let name = "foo";
-            let left_object = WithName::named(name.to_owned(), schema::ObjectType {
-                fields: left_fields.into_iter().map(|(k, v)| (k, schema::ObjectField{r#type: v, description: None})).collect(),
+            let left_object = WithName::named(name.into(), schema::ObjectType {
+                fields: left_fields.into_iter().map(|(k, v)| (k.into(), schema::ObjectField{r#type: v, description: None})).collect(),
                 description: None
             });
-            let right_object = WithName::named(name.to_owned(), schema::ObjectType {
-                fields: right_fields.into_iter().map(|(k, v)| (k, schema::ObjectField{r#type: v, description: None})).collect(),
+            let right_object = WithName::named(name.into(), schema::ObjectType {
+                fields: right_fields.into_iter().map(|(k, v)| (k.into(), schema::ObjectField{r#type: v, description: None})).collect(),
                 description: None
             });
             let result = unify_object_type(left_object, right_object);
 
             for field in result.value.named_fields() {
                 // Any fields not shared between the two input types should be nullable.
-                if !shared.contains_key(field.name) {
+                if !shared.contains_key(field.name.as_str()) {
                     assert!(is_nullable(&field.value.r#type), "Found a non-shared field that is not nullable")
                 }
             }
 
             // All input fields must appear in the result.
-            let fields: HashSet<String> = result.value.fields.into_keys().collect();
-            assert!(left.into_keys().chain(right.into_keys()).chain(shared.into_keys()).all(|k| fields.contains(&k)),
+            let fields: HashSet<ndc_models::FieldName> = result.value.fields.into_keys().collect();
+            assert!(left.into_keys().chain(right.into_keys()).chain(shared.into_keys()).all(|k| fields.contains(&ndc_models::FieldName::from(k))),
                 "Missing field in result type")
         }
     }
diff --git a/crates/cli/src/introspection/validation_schema.rs b/crates/cli/src/introspection/validation_schema.rs
index 2ff37ce8..78ee7d25 100644
--- a/crates/cli/src/introspection/validation_schema.rs
+++ b/crates/cli/src/introspection/validation_schema.rs
@@ -14,9 +14,9 @@ use mongodb_support::BsonScalarType;
 
 use mongodb_agent_common::interface_types::MongoAgentError;
 
-type Collection = WithName<schema::Collection>;
-type ObjectType = WithName<schema::ObjectType>;
-type ObjectField = WithName<schema::ObjectField>;
+type Collection = WithName<ndc_models::CollectionName, schema::Collection>;
+type ObjectType = WithName<ndc_models::ObjectTypeName, schema::ObjectType>;
+type ObjectField = WithName<ndc_models::FieldName, schema::ObjectField>;
 
 pub async fn get_metadata_from_validation_schema(
     state: &ConnectorState,
@@ -24,7 +24,7 @@ pub async fn get_metadata_from_validation_schema(
     let db = state.database();
     let mut collections_cursor = db.list_collections(None, None).await?;
 
-    let mut schemas: Vec<WithName<Schema>> = vec![];
+    let mut schemas: Vec<WithName<String, Schema>> = vec![];
 
     while let Some(collection_spec) = collections_cursor.try_next().await? {
         let name = &collection_spec.name;
@@ -50,10 +50,10 @@ pub async fn get_metadata_from_validation_schema(
 fn make_collection_schema(
     collection_name: &str,
     validator_schema: &ValidatorSchema,
-) -> WithName<Schema> {
+) -> WithName<String, Schema> {
     let (object_types, collection) = make_collection(collection_name, validator_schema);
     WithName::named(
-        collection.name.clone(),
+        collection.name.to_string(),
         Schema {
             collections: WithName::into_map(vec![collection]),
             object_types: WithName::into_map(object_types),
@@ -71,7 +71,7 @@ fn make_collection(
     let (mut object_type_defs, object_fields) = {
         let type_prefix = format!("{collection_name}_");
         let id_field = WithName::named(
-            "_id",
+            "_id".into(),
             schema::ObjectField {
                 description: Some("primary key _id".to_string()),
                 r#type: Type::Scalar(BsonScalarType::ObjectId),
@@ -82,7 +82,7 @@ fn make_collection(
                 .iter()
                 .map(|prop| make_object_field(&type_prefix, required_labels, prop))
                 .unzip();
-        if !object_fields.iter().any(|info| info.name == "_id") {
+        if !object_fields.iter().any(|info| info.name == "_id".into()) {
             // There should always be an _id field, so add it unless it was already specified in
             // the validator.
             object_fields.push(id_field);
@@ -91,7 +91,7 @@ fn make_collection(
     };
 
     let collection_type = WithName::named(
-        collection_name,
+        collection_name.into(),
         schema::ObjectType {
             description: Some(format!("Object type for collection {collection_name}")),
             fields: WithName::into_map(object_fields),
@@ -101,10 +101,10 @@ fn make_collection(
     object_type_defs.push(collection_type);
 
     let collection_info = WithName::named(
-        collection_name,
+        collection_name.into(),
         schema::Collection {
             description: validator_schema.description.clone(),
-            r#type: collection_name.to_string(),
+            r#type: collection_name.into(),
         },
     );
 
@@ -122,7 +122,7 @@ fn make_object_field(
     let (collected_otds, field_type) = make_field_type(&object_type_name, prop_schema);
 
     let object_field = WithName::named(
-        prop_name.clone(),
+        prop_name.to_owned().into(),
         schema::ObjectField {
             description,
             r#type: maybe_nullable(field_type, !required_labels.contains(prop_name)),
@@ -160,7 +160,7 @@ fn make_field_type(object_type_name: &str, prop_schema: &Property) -> (Vec<Objec
                 .unzip();
 
             let object_type_definition = WithName::named(
-                object_type_name.to_string(),
+                object_type_name.into(),
                 schema::ObjectType {
                     description: Some("generated from MongoDB validation schema".to_string()),
                     fields: WithName::into_map(otd_fields),
diff --git a/crates/configuration/src/configuration.rs b/crates/configuration/src/configuration.rs
index e5be5ed3..5ac8131e 100644
--- a/crates/configuration/src/configuration.rs
+++ b/crates/configuration/src/configuration.rs
@@ -17,7 +17,7 @@ pub struct Configuration {
     /// Tracked collections from the configured MongoDB database. This includes real collections as
     /// well as virtual collections defined by native queries using
     /// [NativeQueryRepresentation::Collection] representation.
-    pub collections: BTreeMap<String, ndc::CollectionInfo>,
+    pub collections: BTreeMap<ndc::CollectionName, ndc::CollectionInfo>,
 
     /// Functions are based on native queries using [NativeQueryRepresentation::Function]
     /// representation.
@@ -26,17 +26,17 @@ pub struct Configuration {
     /// responses they are separate concepts. So we want a set of [CollectionInfo] values for
     /// functions for query processing, and we want it separate from `collections` for the schema
     /// response.
-    pub functions: BTreeMap<String, (ndc::FunctionInfo, ndc::CollectionInfo)>,
+    pub functions: BTreeMap<ndc::FunctionName, (ndc::FunctionInfo, ndc::CollectionInfo)>,
 
     /// Procedures are based on native mutations.
-    pub procedures: BTreeMap<String, ndc::ProcedureInfo>,
+    pub procedures: BTreeMap<ndc::ProcedureName, ndc::ProcedureInfo>,
 
     /// Native mutations allow arbitrary MongoDB commands where types of results are specified via
     /// user configuration.
-    pub native_mutations: BTreeMap<String, NativeMutation>,
+    pub native_mutations: BTreeMap<ndc::ProcedureName, NativeMutation>,
 
     /// Native queries allow arbitrary aggregation pipelines that can be included in a query plan.
-    pub native_queries: BTreeMap<String, NativeQuery>,
+    pub native_queries: BTreeMap<ndc::FunctionName, NativeQuery>,
 
     /// Object types defined for this connector include types of documents in each collection,
     /// types for objects inside collection documents, types for native query and native mutation
@@ -45,7 +45,7 @@ pub struct Configuration {
     /// The object types here combine object type defined in files in the `schema/`,
     /// `native_queries/`, and `native_mutations/` subdirectories in the connector configuration
     /// directory.
-    pub object_types: BTreeMap<String, ndc::ObjectType>,
+    pub object_types: BTreeMap<ndc::ObjectTypeName, ndc::ObjectType>,
 
     pub options: ConfigurationOptions,
 }
@@ -53,13 +53,13 @@ pub struct Configuration {
 impl Configuration {
     pub fn validate(
         schema: serialized::Schema,
-        native_mutations: BTreeMap<String, serialized::NativeMutation>,
-        native_queries: BTreeMap<String, serialized::NativeQuery>,
+        native_mutations: BTreeMap<ndc::ProcedureName, serialized::NativeMutation>,
+        native_queries: BTreeMap<ndc::FunctionName, serialized::NativeQuery>,
         options: ConfigurationOptions,
     ) -> anyhow::Result<Self> {
         let object_types_iter = || merge_object_types(&schema, &native_mutations, &native_queries);
         let object_type_errors = {
-            let duplicate_type_names: Vec<&str> = object_types_iter()
+            let duplicate_type_names: Vec<&ndc::TypeName> = object_types_iter()
                 .map(|(name, _)| name.as_ref())
                 .duplicates()
                 .collect();
@@ -68,7 +68,11 @@ impl Configuration {
             } else {
                 Some(anyhow!(
                     "configuration contains multiple definitions for these object type names: {}",
-                    duplicate_type_names.join(", ")
+                    duplicate_type_names
+                        .into_iter()
+                        .map(|tn| tn.to_string())
+                        .collect::<Vec<_>>()
+                        .join(", ")
                 ))
             }
         };
@@ -84,10 +88,10 @@ impl Configuration {
                 )
             });
             let native_query_collections = native_queries.iter().filter_map(
-                |(name, native_query): (&String, &serialized::NativeQuery)| {
+                |(name, native_query): (&ndc::FunctionName, &serialized::NativeQuery)| {
                     if native_query.representation == NativeQueryRepresentation::Collection {
                         Some((
-                            name.to_owned(),
+                            name.as_ref().to_owned(),
                             native_query_to_collection_info(&object_types, name, native_query),
                         ))
                     } else {
@@ -236,9 +240,9 @@ pub struct ConfigurationSerializationOptions {
 
 fn merge_object_types<'a>(
     schema: &'a serialized::Schema,
-    native_mutations: &'a BTreeMap<String, serialized::NativeMutation>,
-    native_queries: &'a BTreeMap<String, serialized::NativeQuery>,
-) -> impl Iterator<Item = (&'a String, &'a schema::ObjectType)> {
+    native_mutations: &'a BTreeMap<ndc::ProcedureName, serialized::NativeMutation>,
+    native_queries: &'a BTreeMap<ndc::FunctionName, serialized::NativeQuery>,
+) -> impl Iterator<Item = (&'a ndc::ObjectTypeName, &'a schema::ObjectType)> {
     let object_types_from_schema = schema.object_types.iter();
     let object_types_from_native_mutations = native_mutations
         .values()
@@ -252,8 +256,8 @@ fn merge_object_types<'a>(
 }
 
 fn collection_to_collection_info(
-    object_types: &BTreeMap<String, schema::ObjectType>,
-    name: String,
+    object_types: &BTreeMap<ndc::ObjectTypeName, schema::ObjectType>,
+    name: ndc::CollectionName,
     collection: schema::Collection,
 ) -> ndc::CollectionInfo {
     let pk_constraint =
@@ -270,19 +274,19 @@ fn collection_to_collection_info(
 }
 
 fn native_query_to_collection_info(
-    object_types: &BTreeMap<String, schema::ObjectType>,
-    name: &str,
+    object_types: &BTreeMap<ndc::ObjectTypeName, schema::ObjectType>,
+    name: &ndc::FunctionName,
     native_query: &serialized::NativeQuery,
 ) -> ndc::CollectionInfo {
     let pk_constraint = get_primary_key_uniqueness_constraint(
         object_types,
-        name,
+        name.as_ref(),
         &native_query.result_document_type,
     );
 
     // TODO: recursively verify that all referenced object types exist
     ndc::CollectionInfo {
-        name: name.to_owned(),
+        name: name.to_owned().into(),
         collection_type: native_query.result_document_type.clone(),
         description: native_query.description.clone(),
         arguments: arguments_to_ndc_arguments(native_query.arguments.clone()),
@@ -292,9 +296,9 @@ fn native_query_to_collection_info(
 }
 
 fn get_primary_key_uniqueness_constraint(
-    object_types: &BTreeMap<String, schema::ObjectType>,
-    name: &str,
-    collection_type: &str,
+    object_types: &BTreeMap<ndc::ObjectTypeName, schema::ObjectType>,
+    name: &ndc::CollectionName,
+    collection_type: &ndc::ObjectTypeName,
 ) -> Option<(String, ndc::UniquenessConstraint)> {
     // Check to make sure our collection's object type contains the _id field
     // If it doesn't (should never happen, all collections need an _id column), don't generate the constraint
@@ -312,8 +316,8 @@ fn get_primary_key_uniqueness_constraint(
 }
 
 fn native_query_to_function_info(
-    object_types: &BTreeMap<String, schema::ObjectType>,
-    name: &str,
+    object_types: &BTreeMap<ndc::ObjectTypeName, schema::ObjectType>,
+    name: &ndc::FunctionName,
     native_query: &serialized::NativeQuery,
 ) -> anyhow::Result<ndc::FunctionInfo> {
     Ok(ndc::FunctionInfo {
@@ -325,9 +329,9 @@ fn native_query_to_function_info(
 }
 
 fn function_result_type(
-    object_types: &BTreeMap<String, schema::ObjectType>,
-    function_name: &str,
-    object_type_name: &str,
+    object_types: &BTreeMap<ndc::ObjectTypeName, schema::ObjectType>,
+    function_name: &ndc::FunctionName,
+    object_type_name: &ndc::ObjectTypeName,
 ) -> anyhow::Result<ndc::Type> {
     let object_type = find_object_type(object_types, object_type_name)?;
     let value_field = object_type.fields.get("__value").ok_or_else(|| {
@@ -338,7 +342,7 @@ fn function_result_type(
 }
 
 fn native_mutation_to_procedure_info(
-    mutation_name: &str,
+    mutation_name: &ndc::ProcedureName,
     mutation: &serialized::NativeMutation,
 ) -> ndc::ProcedureInfo {
     ndc::ProcedureInfo {
@@ -350,8 +354,8 @@ fn native_mutation_to_procedure_info(
 }
 
 fn arguments_to_ndc_arguments(
-    configured_arguments: BTreeMap<String, schema::ObjectField>,
-) -> BTreeMap<String, ndc::ArgumentInfo> {
+    configured_arguments: BTreeMap<ndc::ArgumentName, schema::ObjectField>,
+) -> BTreeMap<ndc::ArgumentName, ndc::ArgumentInfo> {
     configured_arguments
         .into_iter()
         .map(|(name, field)| {
@@ -367,8 +371,8 @@ fn arguments_to_ndc_arguments(
 }
 
 fn find_object_type<'a>(
-    object_types: &'a BTreeMap<String, schema::ObjectType>,
-    object_type_name: &str,
+    object_types: &'a BTreeMap<ndc::ObjectTypeName, schema::ObjectType>,
+    object_type_name: &ndc::ObjectTypeName,
 ) -> anyhow::Result<&'a schema::ObjectType> {
     object_types
         .get(object_type_name)
@@ -387,7 +391,7 @@ mod tests {
         let schema = Schema {
             collections: Default::default(),
             object_types: [(
-                "Album".to_owned(),
+                "Album".to_owned().into(),
                 schema::ObjectType {
                     fields: Default::default(),
                     description: Default::default(),
@@ -397,10 +401,10 @@ mod tests {
             .collect(),
         };
         let native_mutations = [(
-            "hello".to_owned(),
+            "hello".into(),
             serialized::NativeMutation {
                 object_types: [(
-                    "Album".to_owned(),
+                    "Album".to_owned().into(),
                     schema::ObjectType {
                         fields: Default::default(),
                         description: Default::default(),
diff --git a/crates/configuration/src/directory.rs b/crates/configuration/src/directory.rs
index a67e2c24..d94dacd6 100644
--- a/crates/configuration/src/directory.rs
+++ b/crates/configuration/src/directory.rs
@@ -44,7 +44,7 @@ pub async fn read_directory(
 ) -> anyhow::Result<Configuration> {
     let dir = configuration_dir.as_ref();
 
-    let schemas = read_subdir_configs(&dir.join(SCHEMA_DIRNAME))
+    let schemas = read_subdir_configs::<String, Schema>(&dir.join(SCHEMA_DIRNAME))
         .await?
         .unwrap_or_default();
     let schema = schemas.into_values().fold(Schema::default(), Schema::merge);
@@ -75,16 +75,17 @@ pub async fn read_directory(
 /// json and yaml files in the given directory should be parsed as native mutation configurations.
 ///
 /// Assumes that every configuration file has a `name` field.
-async fn read_subdir_configs<T>(subdir: &Path) -> anyhow::Result<Option<BTreeMap<String, T>>>
+async fn read_subdir_configs<N, T>(subdir: &Path) -> anyhow::Result<Option<BTreeMap<N, T>>>
 where
     for<'a> T: Deserialize<'a>,
+    for<'a> N: Ord + ToString + Deserialize<'a>,
 {
     if !(fs::try_exists(subdir).await?) {
         return Ok(None);
     }
 
     let dir_stream = ReadDirStream::new(fs::read_dir(subdir).await?);
-    let configs: Vec<WithName<T>> = dir_stream
+    let configs: Vec<WithName<N, T>> = dir_stream
         .map_err(|err| err.into())
         .try_filter_map(|dir_entry| async move {
             // Permits regular files and symlinks, does not filter out symlinks to directories.
@@ -106,15 +107,15 @@ where
 
             Ok(format_option.map(|format| (path, format)))
         })
-        .and_then(
-            |(path, format)| async move { parse_config_file::<WithName<T>>(path, format).await },
-        )
+        .and_then(|(path, format)| async move {
+            parse_config_file::<WithName<N, T>>(path, format).await
+        })
         .try_collect()
         .await?;
 
     let duplicate_names = configs
         .iter()
-        .map(|c| c.name.as_ref())
+        .map(|c| c.name.to_string())
         .duplicates()
         .collect::<Vec<_>>();
 
@@ -174,7 +175,7 @@ where
     }
 
     for (name, config) in configs {
-        let with_name: WithName<T> = (name.clone(), config).into();
+        let with_name: WithName<String, T> = (name.clone(), config).into();
         write_file(subdir, &name, &with_name).await?;
     }
 
@@ -222,7 +223,7 @@ pub async fn list_existing_schemas(
     let dir = configuration_dir.as_ref();
 
     // TODO: we don't really need to read and parse all the schema files here, just get their names.
-    let schemas = read_subdir_configs::<Schema>(&dir.join(SCHEMA_DIRNAME))
+    let schemas = read_subdir_configs::<_, Schema>(&dir.join(SCHEMA_DIRNAME))
         .await?
         .unwrap_or_default();
 
diff --git a/crates/configuration/src/mongo_scalar_type.rs b/crates/configuration/src/mongo_scalar_type.rs
index 9eb606f6..9641ce9f 100644
--- a/crates/configuration/src/mongo_scalar_type.rs
+++ b/crates/configuration/src/mongo_scalar_type.rs
@@ -15,19 +15,20 @@ pub enum MongoScalarType {
 }
 
 impl MongoScalarType {
-    pub fn lookup_scalar_type(name: &str) -> Option<Self> {
+    pub fn lookup_scalar_type(name: &ndc_models::ScalarTypeName) -> Option<Self> {
         Self::try_from(name).ok()
     }
 }
 
-impl TryFrom<&str> for MongoScalarType {
+impl TryFrom<&ndc_models::ScalarTypeName> for MongoScalarType {
     type Error = QueryPlanError;
 
-    fn try_from(name: &str) -> Result<Self, Self::Error> {
-        if name == EXTENDED_JSON_TYPE_NAME {
+    fn try_from(name: &ndc_models::ScalarTypeName) -> Result<Self, Self::Error> {
+        let name_str = name.to_string();
+        if name_str == EXTENDED_JSON_TYPE_NAME {
             Ok(MongoScalarType::ExtendedJSON)
         } else {
-            let t = BsonScalarType::from_bson_name(name)
+            let t = BsonScalarType::from_bson_name(&name_str)
                 .map_err(|_| QueryPlanError::UnknownScalarType(name.to_owned()))?;
             Ok(MongoScalarType::Bson(t))
         }
diff --git a/crates/configuration/src/native_mutation.rs b/crates/configuration/src/native_mutation.rs
index 5821130a..436673f2 100644
--- a/crates/configuration/src/native_mutation.rs
+++ b/crates/configuration/src/native_mutation.rs
@@ -17,7 +17,7 @@ use crate::{serialized, MongoScalarType};
 #[derive(Clone, Debug)]
 pub struct NativeMutation {
     pub result_type: plan::Type<MongoScalarType>,
-    pub arguments: BTreeMap<String, plan::Type<MongoScalarType>>,
+    pub arguments: BTreeMap<ndc::ArgumentName, plan::Type<MongoScalarType>>,
     pub command: bson::Document,
     pub selection_criteria: Option<SelectionCriteria>,
     pub description: Option<String>,
@@ -25,7 +25,7 @@ pub struct NativeMutation {
 
 impl NativeMutation {
     pub fn from_serialized(
-        object_types: &BTreeMap<String, ndc::ObjectType>,
+        object_types: &BTreeMap<ndc::ObjectTypeName, ndc::ObjectType>,
         input: serialized::NativeMutation,
     ) -> Result<NativeMutation, QueryPlanError> {
         let arguments = input
diff --git a/crates/configuration/src/native_query.rs b/crates/configuration/src/native_query.rs
index e057a90f..3eea44a2 100644
--- a/crates/configuration/src/native_query.rs
+++ b/crates/configuration/src/native_query.rs
@@ -19,16 +19,16 @@ use crate::{serialized, MongoScalarType};
 #[derive(Clone, Debug)]
 pub struct NativeQuery {
     pub representation: NativeQueryRepresentation,
-    pub input_collection: Option<String>,
-    pub arguments: BTreeMap<String, plan::Type<MongoScalarType>>,
-    pub result_document_type: String,
+    pub input_collection: Option<ndc::CollectionName>,
+    pub arguments: BTreeMap<ndc::ArgumentName, plan::Type<MongoScalarType>>,
+    pub result_document_type: ndc::ObjectTypeName,
     pub pipeline: Vec<bson::Document>,
     pub description: Option<String>,
 }
 
 impl NativeQuery {
     pub fn from_serialized(
-        object_types: &BTreeMap<String, ndc::ObjectType>,
+        object_types: &BTreeMap<ndc::ObjectTypeName, ndc::ObjectType>,
         input: serialized::NativeQuery,
     ) -> Result<NativeQuery, QueryPlanError> {
         let arguments = input
diff --git a/crates/configuration/src/schema/mod.rs b/crates/configuration/src/schema/mod.rs
index d69a658e..465fe724 100644
--- a/crates/configuration/src/schema/mod.rs
+++ b/crates/configuration/src/schema/mod.rs
@@ -12,7 +12,7 @@ use crate::{WithName, WithNameRef};
 pub struct Collection {
     /// The name of a type declared in `objectTypes` that describes the fields of this collection.
     /// The type name may be the same as the collection name.
-    pub r#type: String,
+    pub r#type: ndc_models::ObjectTypeName,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub description: Option<String>,
 }
@@ -69,13 +69,15 @@ impl From<Type> for ndc_models::Type {
                 // ExtendedJSON can respresent any BSON value, including null, so it is always nullable
                 Type::ExtendedJSON => ndc_models::Type::Nullable {
                     underlying_type: Box::new(ndc_models::Type::Named {
-                        name: mongodb_support::EXTENDED_JSON_TYPE_NAME.to_owned(),
+                        name: mongodb_support::EXTENDED_JSON_TYPE_NAME.to_owned().into(),
                     }),
                 },
                 Type::Scalar(t) => ndc_models::Type::Named {
-                    name: t.graphql_name().to_owned(),
+                    name: t.graphql_name().to_owned().into(),
+                },
+                Type::Object(t) => ndc_models::Type::Named {
+                    name: t.clone().into(),
                 },
-                Type::Object(t) => ndc_models::Type::Named { name: t.clone() },
                 Type::ArrayOf(t) => ndc_models::Type::Array {
                     element_type: Box::new(map_normalized_type(*t)),
                 },
@@ -91,19 +93,23 @@ impl From<Type> for ndc_models::Type {
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
 #[serde(rename_all = "camelCase")]
 pub struct ObjectType {
-    pub fields: BTreeMap<String, ObjectField>,
+    pub fields: BTreeMap<ndc_models::FieldName, ObjectField>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub description: Option<String>,
 }
 
 impl ObjectType {
-    pub fn named_fields(&self) -> impl Iterator<Item = WithNameRef<'_, ObjectField>> {
+    pub fn named_fields(
+        &self,
+    ) -> impl Iterator<Item = WithNameRef<'_, ndc_models::FieldName, ObjectField>> {
         self.fields
             .iter()
             .map(|(name, field)| WithNameRef::named(name, field))
     }
 
-    pub fn into_named_fields(self) -> impl Iterator<Item = WithName<ObjectField>> {
+    pub fn into_named_fields(
+        self,
+    ) -> impl Iterator<Item = WithName<ndc_models::FieldName, ObjectField>> {
         self.fields
             .into_iter()
             .map(|(name, field)| WithName::named(name, field))
diff --git a/crates/configuration/src/serialized/native_mutation.rs b/crates/configuration/src/serialized/native_mutation.rs
index 9bc6c5d2..cd153171 100644
--- a/crates/configuration/src/serialized/native_mutation.rs
+++ b/crates/configuration/src/serialized/native_mutation.rs
@@ -17,7 +17,7 @@ pub struct NativeMutation {
     /// be merged with the definitions in `schema.json`. This allows you to maintain hand-written
     /// types for native mutations without having to edit a generated `schema.json` file.
     #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
-    pub object_types: BTreeMap<String, ObjectType>,
+    pub object_types: BTreeMap<ndc_models::ObjectTypeName, ObjectType>,
 
     /// Type of data returned by the mutation. You may reference object types defined in the
     /// `object_types` list in this definition, or you may reference object types from
@@ -30,7 +30,7 @@ pub struct NativeMutation {
     /// Argument values are standard JSON mapped from GraphQL input types, not Extended JSON.
     /// Values will be converted to BSON according to the types specified here.
     #[serde(default)]
-    pub arguments: BTreeMap<String, ObjectField>,
+    pub arguments: BTreeMap<ndc_models::ArgumentName, ObjectField>,
 
     /// Command to run via MongoDB's `runCommand` API. For details on how to write commands see
     /// https://www.mongodb.com/docs/manual/reference/method/db.runCommand/
diff --git a/crates/configuration/src/serialized/native_query.rs b/crates/configuration/src/serialized/native_query.rs
index d2042384..11ff4b87 100644
--- a/crates/configuration/src/serialized/native_query.rs
+++ b/crates/configuration/src/serialized/native_query.rs
@@ -35,7 +35,7 @@ pub struct NativeQuery {
 
     /// Use `input_collection` when you want to start an aggregation pipeline off of the specified
     /// `input_collection` db.<input_collection>.aggregate.
-    pub input_collection: Option<String>,
+    pub input_collection: Option<ndc_models::CollectionName>,
 
     /// Arguments to be supplied for each query invocation. These will be available to the given
     /// pipeline as variables. For information about variables in MongoDB aggregation expressions
@@ -44,7 +44,7 @@ pub struct NativeQuery {
     /// Argument values are standard JSON mapped from GraphQL input types, not Extended JSON.
     /// Values will be converted to BSON according to the types specified here.
     #[serde(default)]
-    pub arguments: BTreeMap<String, ObjectField>,
+    pub arguments: BTreeMap<ndc_models::ArgumentName, ObjectField>,
 
     /// The name of an object type that describes documents produced by the given pipeline. MongoDB
     /// aggregation pipelines always produce a list of documents. This type describes the type of
@@ -52,13 +52,13 @@ pub struct NativeQuery {
     ///
     /// You may reference object types defined in the `object_types` list in this definition, or
     /// you may reference object types from `schema.json`.
-    pub result_document_type: String,
+    pub result_document_type: ndc_models::ObjectTypeName,
 
     /// You may define object types here to reference in `result_type`. Any types defined here will
     /// be merged with the definitions in `schema.json`. This allows you to maintain hand-written
     /// types for native queries without having to edit a generated `schema.json` file.
     #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
-    pub object_types: BTreeMap<String, ObjectType>,
+    pub object_types: BTreeMap<ndc_models::ObjectTypeName, ObjectType>,
 
     /// Pipeline to include in MongoDB queries. For details on how to write an aggregation pipeline
     /// see https://www.mongodb.com/docs/manual/core/aggregation-pipeline/
diff --git a/crates/configuration/src/serialized/schema.rs b/crates/configuration/src/serialized/schema.rs
index c3143c81..d9859574 100644
--- a/crates/configuration/src/serialized/schema.rs
+++ b/crates/configuration/src/serialized/schema.rs
@@ -12,31 +12,39 @@ use crate::{
 #[serde(rename_all = "camelCase")]
 pub struct Schema {
     #[serde(default)]
-    pub collections: BTreeMap<String, Collection>,
+    pub collections: BTreeMap<ndc_models::CollectionName, Collection>,
     #[serde(default)]
-    pub object_types: BTreeMap<String, ObjectType>,
+    pub object_types: BTreeMap<ndc_models::ObjectTypeName, ObjectType>,
 }
 
 impl Schema {
-    pub fn into_named_collections(self) -> impl Iterator<Item = WithName<Collection>> {
+    pub fn into_named_collections(
+        self,
+    ) -> impl Iterator<Item = WithName<ndc_models::CollectionName, Collection>> {
         self.collections
             .into_iter()
             .map(|(name, field)| WithName::named(name, field))
     }
 
-    pub fn into_named_object_types(self) -> impl Iterator<Item = WithName<ObjectType>> {
+    pub fn into_named_object_types(
+        self,
+    ) -> impl Iterator<Item = WithName<ndc_models::ObjectTypeName, ObjectType>> {
         self.object_types
             .into_iter()
             .map(|(name, field)| WithName::named(name, field))
     }
 
-    pub fn named_collections(&self) -> impl Iterator<Item = WithNameRef<'_, Collection>> {
+    pub fn named_collections(
+        &self,
+    ) -> impl Iterator<Item = WithNameRef<'_, ndc_models::CollectionName, Collection>> {
         self.collections
             .iter()
             .map(|(name, field)| WithNameRef::named(name, field))
     }
 
-    pub fn named_object_types(&self) -> impl Iterator<Item = WithNameRef<'_, ObjectType>> {
+    pub fn named_object_types(
+        &self,
+    ) -> impl Iterator<Item = WithNameRef<'_, ndc_models::ObjectTypeName, ObjectType>> {
         self.object_types
             .iter()
             .map(|(name, field)| WithNameRef::named(name, field))
diff --git a/crates/configuration/src/with_name.rs b/crates/configuration/src/with_name.rs
index 13332908..85afbfdd 100644
--- a/crates/configuration/src/with_name.rs
+++ b/crates/configuration/src/with_name.rs
@@ -4,16 +4,16 @@ use serde::{Deserialize, Serialize};
 /// deserialize to a map where names are stored as map keys. But in serialized form the name may be
 /// an inline field.
 #[derive(Clone, Debug, Hash, Eq, PartialEq, Ord, PartialOrd, Deserialize, Serialize)]
-pub struct WithName<T> {
-    pub name: String,
+pub struct WithName<N, T> {
+    pub name: N,
     #[serde(flatten)]
     pub value: T,
 }
 
-impl<T> WithName<T> {
-    pub fn into_map<Map>(values: impl IntoIterator<Item = WithName<T>>) -> Map
+impl<N, T> WithName<N, T> {
+    pub fn into_map<Map>(values: impl IntoIterator<Item = WithName<N, T>>) -> Map
     where
-        Map: FromIterator<(String, T)>,
+        Map: FromIterator<(N, T)>,
     {
         values
             .into_iter()
@@ -21,61 +21,61 @@ impl<T> WithName<T> {
             .collect::<Map>()
     }
 
-    pub fn into_name_value_pair(self) -> (String, T) {
+    pub fn into_name_value_pair(self) -> (N, T) {
         (self.name, self.value)
     }
 
-    pub fn named(name: impl ToString, value: T) -> Self {
-        WithName {
-            name: name.to_string(),
-            value,
-        }
+    pub fn named(name: N, value: T) -> Self {
+        WithName { name, value }
     }
 
-    pub fn as_ref<R>(&self) -> WithNameRef<'_, R>
+    pub fn as_ref<RN, RT>(&self) -> WithNameRef<'_, RN, RT>
     where
-        T: AsRef<R>,
+        N: AsRef<RN>,
+        T: AsRef<RT>,
     {
-        WithNameRef::named(&self.name, self.value.as_ref())
+        WithNameRef::named(self.name.as_ref(), self.value.as_ref())
     }
 }
 
-impl<T> From<WithName<T>> for (String, T) {
-    fn from(value: WithName<T>) -> Self {
+impl<N, T> From<WithName<N, T>> for (N, T) {
+    fn from(value: WithName<N, T>) -> Self {
         value.into_name_value_pair()
     }
 }
 
-impl<T> From<(String, T)> for WithName<T> {
-    fn from((name, value): (String, T)) -> Self {
+impl<N, T> From<(N, T)> for WithName<N, T> {
+    fn from((name, value): (N, T)) -> Self {
         WithName::named(name, value)
     }
 }
 
 #[derive(Clone, Copy, Debug, Hash, Eq, PartialEq, Ord, PartialOrd)]
-pub struct WithNameRef<'a, T> {
-    pub name: &'a str,
+pub struct WithNameRef<'a, N, T> {
+    pub name: &'a N,
     pub value: &'a T,
 }
 
-impl<'a, T> WithNameRef<'a, T> {
-    pub fn named<'b>(name: &'b str, value: &'b T) -> WithNameRef<'b, T> {
+impl<'a, N, T> WithNameRef<'a, N, T> {
+    pub fn named<'b>(name: &'b N, value: &'b T) -> WithNameRef<'b, N, T> {
         WithNameRef { name, value }
     }
 
-    pub fn to_owned<R>(&self) -> WithName<R>
+    pub fn to_owned<RN, RT>(&self) -> WithName<RN, RT>
     where
-        T: ToOwned<Owned = R>,
+        N: ToOwned<Owned = RN>,
+        T: ToOwned<Owned = RT>,
     {
         WithName::named(self.name.to_owned(), self.value.to_owned())
     }
 }
 
-impl<'a, T, R> From<&'a WithName<T>> for WithNameRef<'a, R>
+impl<'a, N, T, RN, RT> From<&'a WithName<N, T>> for WithNameRef<'a, RN, RT>
 where
-    T: AsRef<R>,
+    N: AsRef<RN>,
+    T: AsRef<RT>,
 {
-    fn from(value: &'a WithName<T>) -> Self {
+    fn from(value: &'a WithName<N, T>) -> Self {
         value.as_ref()
     }
 }
diff --git a/crates/mongodb-agent-common/src/aggregation_function.rs b/crates/mongodb-agent-common/src/aggregation_function.rs
index bc1cc264..54cb0c0f 100644
--- a/crates/mongodb-agent-common/src/aggregation_function.rs
+++ b/crates/mongodb-agent-common/src/aggregation_function.rs
@@ -28,7 +28,7 @@ impl AggregationFunction {
         all::<AggregationFunction>()
             .find(|variant| variant.graphql_name() == s)
             .ok_or(QueryPlanError::UnknownAggregateFunction {
-                aggregate_function: s.to_owned(),
+                aggregate_function: s.to_owned().into(),
             })
     }
 
diff --git a/crates/mongodb-agent-common/src/comparison_function.rs b/crates/mongodb-agent-common/src/comparison_function.rs
index 881c0d61..09d288ed 100644
--- a/crates/mongodb-agent-common/src/comparison_function.rs
+++ b/crates/mongodb-agent-common/src/comparison_function.rs
@@ -54,7 +54,9 @@ impl ComparisonFunction {
     pub fn from_graphql_name(s: &str) -> Result<Self, QueryPlanError> {
         all::<ComparisonFunction>()
             .find(|variant| variant.graphql_name() == s)
-            .ok_or(QueryPlanError::UnknownComparisonOperator(s.to_owned()))
+            .ok_or(QueryPlanError::UnknownComparisonOperator(
+                s.to_owned().into(),
+            ))
     }
 
     /// Produce a MongoDB expression for use in a match query that applies this function to the given operands.
diff --git a/crates/mongodb-agent-common/src/explain.rs b/crates/mongodb-agent-common/src/explain.rs
index 8c924f76..4e556521 100644
--- a/crates/mongodb-agent-common/src/explain.rs
+++ b/crates/mongodb-agent-common/src/explain.rs
@@ -24,7 +24,7 @@ pub async fn explain_query(
 
     let target = QueryTarget::for_request(config, &query_plan);
     let aggregate_target = match (target.input_collection(), query_plan.has_variables()) {
-        (Some(collection_name), false) => Bson::String(collection_name.to_owned()),
+        (Some(collection_name), false) => Bson::String(collection_name.to_string()),
         _ => Bson::Int32(1),
     };
 
diff --git a/crates/mongodb-agent-common/src/interface_types/mod.rs b/crates/mongodb-agent-common/src/interface_types/mod.rs
index bd9e5d35..13be2c05 100644
--- a/crates/mongodb-agent-common/src/interface_types/mod.rs
+++ b/crates/mongodb-agent-common/src/interface_types/mod.rs
@@ -1,3 +1,3 @@
 mod mongo_agent_error;
 
-pub use self::mongo_agent_error::MongoAgentError;
+pub use self::mongo_agent_error::{ErrorResponse, MongoAgentError};
diff --git a/crates/mongodb-agent-common/src/mongo_query_plan/mod.rs b/crates/mongodb-agent-common/src/mongo_query_plan/mod.rs
index 203bc7d0..57f54cdc 100644
--- a/crates/mongodb-agent-common/src/mongo_query_plan/mod.rs
+++ b/crates/mongodb-agent-common/src/mongo_query_plan/mod.rs
@@ -21,11 +21,11 @@ impl MongoConfiguration {
         self.0.options.serialization_options.extended_json_mode
     }
 
-    pub fn native_queries(&self) -> &BTreeMap<String, NativeQuery> {
+    pub fn native_queries(&self) -> &BTreeMap<ndc::FunctionName, NativeQuery> {
         &self.0.native_queries
     }
 
-    pub fn native_mutations(&self) -> &BTreeMap<String, NativeMutation> {
+    pub fn native_mutations(&self) -> &BTreeMap<ndc::ProcedureName, NativeMutation> {
         &self.0.native_mutations
     }
 }
@@ -37,16 +37,16 @@ impl ConnectorTypes for MongoConfiguration {
 }
 
 impl QueryContext for MongoConfiguration {
-    fn lookup_scalar_type(type_name: &str) -> Option<Self::ScalarType> {
+    fn lookup_scalar_type(type_name: &ndc::ScalarTypeName) -> Option<Self::ScalarType> {
         type_name.try_into().ok()
     }
 
     fn lookup_aggregation_function(
         &self,
         input_type: &Type,
-        function_name: &str,
+        function_name: &ndc::AggregateFunctionName,
     ) -> Result<(Self::AggregateFunction, &ndc::AggregateFunctionDefinition), QueryPlanError> {
-        let function = AggregationFunction::from_graphql_name(function_name)?;
+        let function = AggregationFunction::from_graphql_name(function_name.as_str())?;
         let definition = scalar_type_name(input_type)
             .and_then(|name| SCALAR_TYPES.get(name))
             .and_then(|scalar_type_def| scalar_type_def.aggregate_functions.get(function_name))
@@ -59,12 +59,12 @@ impl QueryContext for MongoConfiguration {
     fn lookup_comparison_operator(
         &self,
         left_operand_type: &Type,
-        operator_name: &str,
+        operator_name: &ndc::ComparisonOperatorName,
     ) -> Result<(Self::ComparisonOperator, &ndc::ComparisonOperatorDefinition), QueryPlanError>
     where
         Self: Sized,
     {
-        let operator = ComparisonFunction::from_graphql_name(operator_name)?;
+        let operator = ComparisonFunction::from_graphql_name(operator_name.as_str())?;
         let definition = scalar_type_name(left_operand_type)
             .and_then(|name| SCALAR_TYPES.get(name))
             .and_then(|scalar_type_def| scalar_type_def.comparison_operators.get(operator_name))
@@ -72,19 +72,19 @@ impl QueryContext for MongoConfiguration {
         Ok((operator, definition))
     }
 
-    fn collections(&self) -> &BTreeMap<String, ndc::CollectionInfo> {
+    fn collections(&self) -> &BTreeMap<ndc::CollectionName, ndc::CollectionInfo> {
         &self.0.collections
     }
 
-    fn functions(&self) -> &BTreeMap<String, (ndc::FunctionInfo, ndc::CollectionInfo)> {
+    fn functions(&self) -> &BTreeMap<ndc::FunctionName, (ndc::FunctionInfo, ndc::CollectionInfo)> {
         &self.0.functions
     }
 
-    fn object_types(&self) -> &BTreeMap<String, ndc::ObjectType> {
+    fn object_types(&self) -> &BTreeMap<ndc::ObjectTypeName, ndc::ObjectType> {
         &self.0.object_types
     }
 
-    fn procedures(&self) -> &BTreeMap<String, ndc::ProcedureInfo> {
+    fn procedures(&self) -> &BTreeMap<ndc::ProcedureName, ndc::ProcedureInfo> {
         &self.0.procedures
     }
 }
diff --git a/crates/mongodb-agent-common/src/mongodb/selection.rs b/crates/mongodb-agent-common/src/mongodb/selection.rs
index cc7d7721..4c8c2ee8 100644
--- a/crates/mongodb-agent-common/src/mongodb/selection.rs
+++ b/crates/mongodb-agent-common/src/mongodb/selection.rs
@@ -39,11 +39,11 @@ impl Selection {
 
 fn from_query_request_helper(
     parent_columns: &[&str],
-    field_selection: &IndexMap<String, Field>,
+    field_selection: &IndexMap<ndc_models::FieldName, Field>,
 ) -> Result<Document, MongoAgentError> {
     field_selection
         .iter()
-        .map(|(key, value)| Ok((key.into(), selection_for_field(parent_columns, value)?)))
+        .map(|(key, value)| Ok((key.to_string(), selection_for_field(parent_columns, value)?)))
         .collect()
 }
 
@@ -73,7 +73,7 @@ fn selection_for_field(parent_columns: &[&str], field: &Field) -> Result<Bson, M
             fields: Some(NestedField::Object(NestedObject { fields })),
             ..
         } => {
-            let nested_parent_columns = append_to_path(parent_columns, column);
+            let nested_parent_columns = append_to_path(parent_columns, column.as_str());
             let nested_parent_col_path = format!("${}", nested_parent_columns.join("."));
             let nested_selection = from_query_request_helper(&nested_parent_columns, fields)?;
             Ok(doc! {"$cond": {"if": nested_parent_col_path, "then": nested_selection, "else": Bson::Null}}.into())
@@ -85,7 +85,11 @@ fn selection_for_field(parent_columns: &[&str], field: &Field) -> Result<Bson, M
                     fields: nested_field,
                 })),
             ..
-        } => selection_for_array(&append_to_path(parent_columns, column), nested_field, 0),
+        } => selection_for_array(
+            &append_to_path(parent_columns, column.as_str()),
+            nested_field,
+            0,
+        ),
         Field::Relationship {
             relationship,
             aggregates,
@@ -100,7 +104,10 @@ fn selection_for_field(parent_columns: &[&str], field: &Field) -> Result<Bson, M
                 fields
                     .iter()
                     .map(|(field_name, _)| {
-                        (field_name.to_owned(), format!("$$this.{field_name}").into())
+                        (
+                            field_name.to_string(),
+                            format!("$$this.{field_name}").into(),
+                        )
                     })
                     .collect()
             });
@@ -110,7 +117,7 @@ fn selection_for_field(parent_columns: &[&str], field: &Field) -> Result<Bson, M
                     .iter()
                     .map(|(aggregate_name, _)| {
                         (
-                            aggregate_name.to_owned(),
+                            aggregate_name.to_string(),
                             format!("$$row_set.aggregates.{aggregate_name}").into(),
                         )
                     })
@@ -131,7 +138,7 @@ fn selection_for_field(parent_columns: &[&str], field: &Field) -> Result<Bson, M
 
                 Ok(doc! {
                     "$let": {
-                        "vars": { "row_set": { "$first": get_field(relationship) } },
+                        "vars": { "row_set": { "$first": get_field(relationship.as_str()) } },
                         "in": new_row_set,
                     }
                 }
@@ -140,7 +147,7 @@ fn selection_for_field(parent_columns: &[&str], field: &Field) -> Result<Bson, M
                 Ok(doc! {
                     "rows": {
                         "$map": {
-                            "input": get_field(relationship),
+                            "input": get_field(relationship.as_str()),
                             "in": field_selection,
                         }
                     }
diff --git a/crates/mongodb-agent-common/src/procedure/error.rs b/crates/mongodb-agent-common/src/procedure/error.rs
index 45a5ba56..bff2afab 100644
--- a/crates/mongodb-agent-common/src/procedure/error.rs
+++ b/crates/mongodb-agent-common/src/procedure/error.rs
@@ -9,10 +9,10 @@ pub enum ProcedureError {
     ExecutionError(#[from] mongodb::error::Error),
 
     #[error("a required argument was not provided, \"{0}\"")]
-    MissingArgument(String),
+    MissingArgument(ndc_models::ArgumentName),
 
     #[error("found a non-string argument, {0}, in a string context - if you want to use a non-string argument it must be the only thing in the string with no white space around the curly braces")]
-    NonStringInStringContext(String),
+    NonStringInStringContext(ndc_models::ArgumentName),
 
     #[error("object keys must be strings, but got: \"{0}\"")]
     NonStringKey(Bson),
diff --git a/crates/mongodb-agent-common/src/procedure/interpolated_command.rs b/crates/mongodb-agent-common/src/procedure/interpolated_command.rs
index b3e555c4..f04dde4c 100644
--- a/crates/mongodb-agent-common/src/procedure/interpolated_command.rs
+++ b/crates/mongodb-agent-common/src/procedure/interpolated_command.rs
@@ -10,7 +10,7 @@ type Result<T> = std::result::Result<T, ProcedureError>;
 /// Parse native mutation commands, and interpolate arguments.
 pub fn interpolated_command(
     command: &bson::Document,
-    arguments: &BTreeMap<String, Bson>,
+    arguments: &BTreeMap<ndc_models::ArgumentName, Bson>,
 ) -> Result<bson::Document> {
     let bson = interpolate_helper(&command.into(), arguments)?;
     match bson {
@@ -19,7 +19,10 @@ pub fn interpolated_command(
     }
 }
 
-fn interpolate_helper(command_node: &Bson, arguments: &BTreeMap<String, Bson>) -> Result<Bson> {
+fn interpolate_helper(
+    command_node: &Bson,
+    arguments: &BTreeMap<ndc_models::ArgumentName, Bson>,
+) -> Result<Bson> {
     let result = match command_node {
         Bson::Array(values) => interpolate_array(values.to_vec(), arguments)?.into(),
         Bson::Document(doc) => interpolate_document(doc.clone(), arguments)?.into(),
@@ -30,7 +33,10 @@ fn interpolate_helper(command_node: &Bson, arguments: &BTreeMap<String, Bson>) -
     Ok(result)
 }
 
-fn interpolate_array(values: Vec<Bson>, arguments: &BTreeMap<String, Bson>) -> Result<Vec<Bson>> {
+fn interpolate_array(
+    values: Vec<Bson>,
+    arguments: &BTreeMap<ndc_models::ArgumentName, Bson>,
+) -> Result<Vec<Bson>> {
     values
         .iter()
         .map(|value| interpolate_helper(value, arguments))
@@ -39,7 +45,7 @@ fn interpolate_array(values: Vec<Bson>, arguments: &BTreeMap<String, Bson>) -> R
 
 fn interpolate_document(
     document: bson::Document,
-    arguments: &BTreeMap<String, Bson>,
+    arguments: &BTreeMap<ndc_models::ArgumentName, Bson>,
 ) -> Result<bson::Document> {
     document
         .into_iter()
@@ -68,7 +74,10 @@ fn interpolate_document(
 /// ```
 ///
 /// if the type of the variable `recordId` is `int`.
-fn interpolate_string(string: &str, arguments: &BTreeMap<String, Bson>) -> Result<Bson> {
+fn interpolate_string(
+    string: &str,
+    arguments: &BTreeMap<ndc_models::ArgumentName, Bson>,
+) -> Result<Bson> {
     let parts = parse_native_mutation(string);
     if parts.len() == 1 {
         let mut parts = parts;
@@ -94,7 +103,10 @@ fn interpolate_string(string: &str, arguments: &BTreeMap<String, Bson>) -> Resul
     }
 }
 
-fn resolve_argument(argument_name: &str, arguments: &BTreeMap<String, Bson>) -> Result<Bson> {
+fn resolve_argument(
+    argument_name: &ndc_models::ArgumentName,
+    arguments: &BTreeMap<ndc_models::ArgumentName, Bson>,
+) -> Result<Bson> {
     let argument = arguments
         .get(argument_name)
         .ok_or_else(|| ProcedureError::MissingArgument(argument_name.to_owned()))?;
@@ -107,7 +119,7 @@ enum NativeMutationPart {
     /// A raw text part
     Text(String),
     /// A parameter
-    Parameter(String),
+    Parameter(ndc_models::ArgumentName),
 }
 
 /// Parse a string or key in a native procedure into parts where variables have the syntax
@@ -120,10 +132,10 @@ fn parse_native_mutation(string: &str) -> Vec<NativeMutationPart> {
             None => vec![NativeMutationPart::Text(part.to_string())],
             Some((var, text)) => {
                 if text.is_empty() {
-                    vec![NativeMutationPart::Parameter(var.trim().to_owned())]
+                    vec![NativeMutationPart::Parameter(var.trim().into())]
                 } else {
                     vec![
-                        NativeMutationPart::Parameter(var.trim().to_owned()),
+                        NativeMutationPart::Parameter(var.trim().into()),
                         NativeMutationPart::Text(text.to_string()),
                     ]
                 }
@@ -157,9 +169,9 @@ mod tests {
                 fields: [("ok".into(), Type::Scalar(MongoScalarType::Bson(S::Bool)))].into(),
             }),
             arguments: [
-                ("id".to_owned(), Type::Scalar(MongoScalarType::Bson(S::Int))),
+                ("id".into(), Type::Scalar(MongoScalarType::Bson(S::Int))),
                 (
-                    "name".to_owned(),
+                    "name".into(),
                     Type::Scalar(MongoScalarType::Bson(S::String)),
                 ),
             ]
@@ -176,9 +188,9 @@ mod tests {
         };
 
         let input_arguments = [
-            ("id".to_owned(), Argument::Literal { value: json!(1001) }),
+            ("id".into(), Argument::Literal { value: json!(1001) }),
             (
-                "name".to_owned(),
+                "name".into(),
                 Argument::Literal {
                     value: json!("Regina Spektor"),
                 },
@@ -211,7 +223,7 @@ mod tests {
                 fields: [("ok".into(), Type::Scalar(MongoScalarType::Bson(S::Bool)))].into(),
             }),
             arguments: [(
-                "documents".to_owned(),
+                "documents".into(),
                 Type::ArrayOf(Box::new(Type::Object(ObjectType {
                     name: Some("ArtistInput".into()),
                     fields: [
@@ -237,7 +249,7 @@ mod tests {
         };
 
         let input_arguments = [(
-            "documents".to_owned(),
+            "documents".into(),
             Argument::Literal {
                 value: json!([
                     { "ArtistId": 1001, "Name": "Regina Spektor" } ,
@@ -279,11 +291,11 @@ mod tests {
             }),
             arguments: [
                 (
-                    "prefix".to_owned(),
+                    "prefix".into(),
                     Type::Scalar(MongoScalarType::Bson(S::String)),
                 ),
                 (
-                    "basename".to_owned(),
+                    "basename".into(),
                     Type::Scalar(MongoScalarType::Bson(S::String)),
                 ),
             ]
@@ -298,13 +310,13 @@ mod tests {
 
         let input_arguments = [
             (
-                "prefix".to_owned(),
+                "prefix".into(),
                 Argument::Literal {
                     value: json!("current"),
                 },
             ),
             (
-                "basename".to_owned(),
+                "basename".into(),
                 Argument::Literal {
                     value: json!("some-coll"),
                 },
diff --git a/crates/mongodb-agent-common/src/procedure/mod.rs b/crates/mongodb-agent-common/src/procedure/mod.rs
index 42ec794e..9729b071 100644
--- a/crates/mongodb-agent-common/src/procedure/mod.rs
+++ b/crates/mongodb-agent-common/src/procedure/mod.rs
@@ -18,9 +18,9 @@ pub use self::interpolated_command::interpolated_command;
 /// Encapsulates running arbitrary mongodb commands with interpolated arguments
 #[derive(Clone, Debug)]
 pub struct Procedure<'a> {
-    arguments: BTreeMap<String, serde_json::Value>,
+    arguments: BTreeMap<ndc_models::ArgumentName, serde_json::Value>,
     command: Cow<'a, bson::Document>,
-    parameters: Cow<'a, BTreeMap<String, Type>>,
+    parameters: Cow<'a, BTreeMap<ndc_models::ArgumentName, Type>>,
     result_type: Type,
     selection_criteria: Option<Cow<'a, SelectionCriteria>>,
 }
@@ -28,7 +28,7 @@ pub struct Procedure<'a> {
 impl<'a> Procedure<'a> {
     pub fn from_native_mutation(
         native_mutation: &'a NativeMutation,
-        arguments: BTreeMap<String, serde_json::Value>,
+        arguments: BTreeMap<ndc_models::ArgumentName, serde_json::Value>,
     ) -> Self {
         Procedure {
             arguments,
@@ -58,8 +58,8 @@ impl<'a> Procedure<'a> {
 }
 
 fn interpolate(
-    parameters: &BTreeMap<String, Type>,
-    arguments: BTreeMap<String, serde_json::Value>,
+    parameters: &BTreeMap<ndc_models::ArgumentName, Type>,
+    arguments: BTreeMap<ndc_models::ArgumentName, serde_json::Value>,
     command: &bson::Document,
 ) -> Result<bson::Document, ProcedureError> {
     let arguments = arguments
diff --git a/crates/mongodb-agent-common/src/query/arguments.rs b/crates/mongodb-agent-common/src/query/arguments.rs
index f5889b02..bd8cdb9a 100644
--- a/crates/mongodb-agent-common/src/query/arguments.rs
+++ b/crates/mongodb-agent-common/src/query/arguments.rs
@@ -16,13 +16,13 @@ use super::{
 #[derive(Debug, Error)]
 pub enum ArgumentError {
     #[error("unknown variables or arguments: {}", .0.join(", "))]
-    Excess(Vec<String>),
+    Excess(Vec<ndc_models::ArgumentName>),
 
     #[error("some variables or arguments are invalid:\n{}", format_errors(.0))]
-    Invalid(BTreeMap<String, JsonToBsonError>),
+    Invalid(BTreeMap<ndc_models::ArgumentName, JsonToBsonError>),
 
     #[error("missing variables or arguments: {}", .0.join(", "))]
-    Missing(Vec<String>),
+    Missing(Vec<ndc_models::ArgumentName>),
 }
 
 /// Translate arguments to queries or native queries to BSON according to declared parameter types.
@@ -30,12 +30,15 @@ pub enum ArgumentError {
 /// Checks that all arguments have been provided, and that no arguments have been given that do not
 /// map to declared parameters (no excess arguments).
 pub fn resolve_arguments(
-    parameters: &BTreeMap<String, Type>,
-    mut arguments: BTreeMap<String, Argument>,
-) -> Result<BTreeMap<String, Bson>, ArgumentError> {
+    parameters: &BTreeMap<ndc_models::ArgumentName, Type>,
+    mut arguments: BTreeMap<ndc_models::ArgumentName, Argument>,
+) -> Result<BTreeMap<ndc_models::ArgumentName, Bson>, ArgumentError> {
     validate_no_excess_arguments(parameters, &arguments)?;
 
-    let (arguments, missing): (Vec<(String, Argument, &Type)>, Vec<String>) = parameters
+    let (arguments, missing): (
+        Vec<(ndc_models::ArgumentName, Argument, &Type)>,
+        Vec<ndc_models::ArgumentName>,
+    ) = parameters
         .iter()
         .map(|(name, parameter_type)| {
             if let Some((name, argument)) = arguments.remove_entry(name) {
@@ -49,7 +52,10 @@ pub fn resolve_arguments(
         return Err(ArgumentError::Missing(missing));
     }
 
-    let (resolved, errors): (BTreeMap<String, Bson>, BTreeMap<String, JsonToBsonError>) = arguments
+    let (resolved, errors): (
+        BTreeMap<ndc_models::ArgumentName, Bson>,
+        BTreeMap<ndc_models::ArgumentName, JsonToBsonError>,
+    ) = arguments
         .into_iter()
         .map(|(name, argument, parameter_type)| {
             match argument_to_mongodb_expression(&argument, parameter_type) {
@@ -79,10 +85,10 @@ fn argument_to_mongodb_expression(
 }
 
 pub fn validate_no_excess_arguments<T>(
-    parameters: &BTreeMap<String, Type>,
-    arguments: &BTreeMap<String, T>,
+    parameters: &BTreeMap<ndc_models::ArgumentName, Type>,
+    arguments: &BTreeMap<ndc_models::ArgumentName, T>,
 ) -> Result<(), ArgumentError> {
-    let excess: Vec<String> = arguments
+    let excess: Vec<ndc_models::ArgumentName> = arguments
         .iter()
         .filter_map(|(name, _)| {
             let parameter = parameters.get(name);
@@ -99,7 +105,7 @@ pub fn validate_no_excess_arguments<T>(
     }
 }
 
-fn format_errors(errors: &BTreeMap<String, JsonToBsonError>) -> String {
+fn format_errors(errors: &BTreeMap<ndc_models::ArgumentName, JsonToBsonError>) -> String {
     errors
         .iter()
         .map(|(name, error)| format!("  {name}:\n{}", indent_all_by(4, error.to_string())))
diff --git a/crates/mongodb-agent-common/src/query/column_ref.rs b/crates/mongodb-agent-common/src/query/column_ref.rs
index 5ed7f25c..cd0bef69 100644
--- a/crates/mongodb-agent-common/src/query/column_ref.rs
+++ b/crates/mongodb-agent-common/src/query/column_ref.rs
@@ -82,10 +82,10 @@ pub fn name_from_scope(scope: &Scope) -> Cow<'_, str> {
 
 fn from_path<'a>(
     init: Option<ColumnRef<'a>>,
-    path: impl IntoIterator<Item = &'a String>,
+    path: impl IntoIterator<Item = &'a ndc_models::FieldName>,
 ) -> Option<ColumnRef<'a>> {
     path.into_iter().fold(init, |accum, element| {
-        Some(fold_path_element(accum, element))
+        Some(fold_path_element(accum, element.as_ref()))
     })
 }
 
diff --git a/crates/mongodb-agent-common/src/query/execute_query_request.rs b/crates/mongodb-agent-common/src/query/execute_query_request.rs
index 406b7e20..bf107318 100644
--- a/crates/mongodb-agent-common/src/query/execute_query_request.rs
+++ b/crates/mongodb-agent-common/src/query/execute_query_request.rs
@@ -63,7 +63,7 @@ async fn execute_query_pipeline(
     // another case where we call `db.aggregate` instead of `db.<collection>.aggregate`.
     let documents = match (target.input_collection(), query_plan.has_variables()) {
         (Some(collection_name), false) => {
-            let collection = database.collection(collection_name);
+            let collection = database.collection(collection_name.as_str());
             collect_response_documents(
                 collection
                     .aggregate(pipeline, None)
diff --git a/crates/mongodb-agent-common/src/query/foreach.rs b/crates/mongodb-agent-common/src/query/foreach.rs
index 217019a8..00bf3596 100644
--- a/crates/mongodb-agent-common/src/query/foreach.rs
+++ b/crates/mongodb-agent-common/src/query/foreach.rs
@@ -92,7 +92,7 @@ fn variable_sets_to_bson(
 /// It may be necessary to include a request variable in the MongoDB pipeline multiple times if it
 /// requires different BSON serializations.
 fn variable_to_bson<'a>(
-    name: &'a str,
+    name: &'a ndc_models::VariableName,
     value: &'a serde_json::Value,
     variable_types: impl IntoIterator<Item = &'a Option<Type>> + 'a,
 ) -> impl Iterator<Item = Result<(String, Bson)>> + 'a {
diff --git a/crates/mongodb-agent-common/src/query/make_selector.rs b/crates/mongodb-agent-common/src/query/make_selector.rs
index ea2bf197..f7ddb7da 100644
--- a/crates/mongodb-agent-common/src/query/make_selector.rs
+++ b/crates/mongodb-agent-common/src/query/make_selector.rs
@@ -44,7 +44,7 @@ pub fn make_selector(expr: &Expression) -> Result<Document> {
         } => Ok(match in_collection {
             ExistsInCollection::Related { relationship } => match predicate {
                 Some(predicate) => doc! {
-                    relationship: { "$elemMatch": make_selector(predicate)? }
+                    relationship.to_string(): { "$elemMatch": make_selector(predicate)? }
                 },
                 None => doc! { format!("{relationship}.0"): { "$exists": true } },
             },
@@ -137,10 +137,13 @@ fn make_binary_comparison_selector(
 /// related documents always come as an array, even for object relationships), so we have to wrap
 /// the starting expression with an `$elemMatch` for each relationship that is traversed to reach
 /// the target column.
-fn traverse_relationship_path(path: &[String], mut expression: Document) -> Document {
+fn traverse_relationship_path(
+    path: &[ndc_models::RelationshipName],
+    mut expression: Document,
+) -> Document {
     for path_element in path.iter().rev() {
         expression = doc! {
-            path_element: {
+            path_element.to_string(): {
                 "$elemMatch": expression
             }
         }
@@ -148,7 +151,10 @@ fn traverse_relationship_path(path: &[String], mut expression: Document) -> Docu
     expression
 }
 
-fn variable_to_mongo_expression(variable: &str, value_type: &Type) -> bson::Bson {
+fn variable_to_mongo_expression(
+    variable: &ndc_models::VariableName,
+    value_type: &Type,
+) -> bson::Bson {
     let mongodb_var_name = query_variable_name(variable, value_type);
     format!("$${mongodb_var_name}").into()
 }
@@ -180,7 +186,7 @@ mod tests {
     ) -> anyhow::Result<()> {
         let selector = make_selector(&Expression::BinaryComparisonOperator {
             column: ComparisonTarget::Column {
-                name: "Name".to_owned(),
+                name: "Name".into(),
                 field_path: None,
                 field_type: Type::Scalar(MongoScalarType::Bson(BsonScalarType::String)),
                 path: vec!["Albums".into(), "Tracks".into()],
@@ -213,7 +219,7 @@ mod tests {
     ) -> anyhow::Result<()> {
         let selector = make_selector(&Expression::UnaryComparisonOperator {
             column: ComparisonTarget::Column {
-                name: "Name".to_owned(),
+                name: "Name".into(),
                 field_path: None,
                 field_type: Type::Scalar(MongoScalarType::Bson(BsonScalarType::String)),
                 path: vec!["Albums".into(), "Tracks".into()],
@@ -241,7 +247,7 @@ mod tests {
     fn compares_two_columns() -> anyhow::Result<()> {
         let selector = make_selector(&Expression::BinaryComparisonOperator {
             column: ComparisonTarget::Column {
-                name: "Name".to_owned(),
+                name: "Name".into(),
                 field_path: None,
                 field_type: Type::Scalar(MongoScalarType::Bson(BsonScalarType::String)),
                 path: Default::default(),
@@ -249,7 +255,7 @@ mod tests {
             operator: ComparisonFunction::Equal,
             value: ComparisonValue::Column {
                 column: ComparisonTarget::Column {
-                    name: "Title".to_owned(),
+                    name: "Title".into(),
                     field_path: None,
                     field_type: Type::Scalar(MongoScalarType::Bson(BsonScalarType::String)),
                     path: Default::default(),
@@ -271,7 +277,7 @@ mod tests {
     fn compares_root_collection_column_to_scalar() -> anyhow::Result<()> {
         let selector = make_selector(&Expression::BinaryComparisonOperator {
             column: ComparisonTarget::ColumnInScope {
-                name: "Name".to_owned(),
+                name: "Name".into(),
                 field_path: None,
                 field_type: Type::Scalar(MongoScalarType::Bson(BsonScalarType::String)),
                 scope: Scope::Named("scope_0".to_string()),
@@ -302,7 +308,7 @@ mod tests {
                 binop(
                     "_gt",
                     target!("Milliseconds", relations: [
-                        path_element("Tracks").predicate(
+                        path_element("Tracks".into()).predicate(
                             binop("_eq", target!("Name"), column_value!(root("Title")))
                         ),
                     ]),
diff --git a/crates/mongodb-agent-common/src/query/make_sort.rs b/crates/mongodb-agent-common/src/query/make_sort.rs
index f32e7704..e113da4e 100644
--- a/crates/mongodb-agent-common/src/query/make_sort.rs
+++ b/crates/mongodb-agent-common/src/query/make_sort.rs
@@ -51,14 +51,15 @@ pub fn make_sort(order_by: &OrderBy) -> Result<Document, MongoAgentError> {
 
 // TODO: MDB-159 Replace use of [safe_name] with [ColumnRef].
 fn column_ref_with_path(
-    name: &String,
-    field_path: Option<&[String]>,
-    relation_path: &[String],
+    name: &ndc_models::FieldName,
+    field_path: Option<&[ndc_models::FieldName]>,
+    relation_path: &[ndc_models::RelationshipName],
 ) -> Result<String, MongoAgentError> {
     relation_path
         .iter()
-        .chain(std::iter::once(name))
-        .chain(field_path.into_iter().flatten())
-        .map(|x| safe_name(x))
+        .map(|n| n.as_str())
+        .chain(std::iter::once(name.as_str()))
+        .chain(field_path.into_iter().flatten().map(|n| n.as_str()))
+        .map(safe_name)
         .process_results(|mut iter| iter.join("."))
 }
diff --git a/crates/mongodb-agent-common/src/query/native_query.rs b/crates/mongodb-agent-common/src/query/native_query.rs
index 56ffc4dc..7b976b4f 100644
--- a/crates/mongodb-agent-common/src/query/native_query.rs
+++ b/crates/mongodb-agent-common/src/query/native_query.rs
@@ -31,7 +31,7 @@ pub fn pipeline_for_native_query(
 
 fn make_pipeline(
     native_query: &NativeQuery,
-    arguments: &BTreeMap<String, Argument>,
+    arguments: &BTreeMap<ndc_models::ArgumentName, Argument>,
 ) -> Result<Pipeline, MongoAgentError> {
     let bson_arguments = resolve_arguments(&native_query.arguments, arguments.clone())
         .map_err(ProcedureError::UnresolvableArguments)?;
@@ -75,28 +75,28 @@ mod tests {
             input_collection: None,
             arguments: [
                 (
-                    "filter".to_string(),
+                    "filter".into(),
                     ObjectField {
                         r#type: Type::ExtendedJSON,
                         description: None,
                     },
                 ),
                 (
-                    "queryVector".to_string(),
+                    "queryVector".into(),
                     ObjectField {
                         r#type: Type::ArrayOf(Box::new(Type::Scalar(S::Double))),
                         description: None,
                     },
                 ),
                 (
-                    "numCandidates".to_string(),
+                    "numCandidates".into(),
                     ObjectField {
                         r#type: Type::Scalar(S::Int),
                         description: None,
                     },
                 ),
                 (
-                    "limit".to_string(),
+                    "limit".into(),
                     ObjectField {
                         r#type: Type::Scalar(S::Int),
                         description: None,
@@ -104,35 +104,35 @@ mod tests {
                 ),
             ]
             .into(),
-            result_document_type: "VectorResult".to_owned(),
+            result_document_type: "VectorResult".into(),
             object_types: [(
-                "VectorResult".to_owned(),
+                "VectorResult".into(),
                 ObjectType {
                     description: None,
                     fields: [
                         (
-                            "_id".to_owned(),
+                            "_id".into(),
                             ObjectField {
                                 r#type: Type::Scalar(S::ObjectId),
                                 description: None,
                             },
                         ),
                         (
-                            "title".to_owned(),
+                            "title".into(),
                             ObjectField {
                                 r#type: Type::Scalar(S::String),
                                 description: None,
                             },
                         ),
                         (
-                            "genres".to_owned(),
+                            "genres".into(),
                             ObjectField {
                                 r#type: Type::ArrayOf(Box::new(Type::Scalar(S::String))),
                                 description: None,
                             },
                         ),
                         (
-                            "year".to_owned(),
+                            "year".into(),
                             ObjectField {
                                 r#type: Type::Scalar(S::Int),
                                 description: None,
diff --git a/crates/mongodb-agent-common/src/query/pipeline.rs b/crates/mongodb-agent-common/src/query/pipeline.rs
index 745a608c..a7fb3868 100644
--- a/crates/mongodb-agent-common/src/query/pipeline.rs
+++ b/crates/mongodb-agent-common/src/query/pipeline.rs
@@ -118,9 +118,10 @@ pub fn pipeline_for_fields_facet(
         // Queries higher up the chain might need to reference relationships from this query. So we
         // forward relationship arrays if this is not the top-level query.
         for relationship_key in relationships.keys() {
-            selection
-                .0
-                .insert(relationship_key.to_owned(), get_field(relationship_key));
+            selection.0.insert(
+                relationship_key.to_owned(),
+                get_field(relationship_key.as_str()),
+            );
         }
     }
 
@@ -153,7 +154,7 @@ fn facet_pipelines_for_query(
         .flatten()
         .map(|(key, aggregate)| {
             Ok((
-                key.clone(),
+                key.to_string(),
                 pipeline_for_aggregate(aggregate.clone(), *aggregates_limit)?,
             ))
         })
@@ -176,7 +177,7 @@ fn facet_pipelines_for_query(
             let value_expr = doc! {
                 "$getField": {
                     "field": RESULT_FIELD, // evaluates to the value of this field
-                    "input": { "$first": get_field(key) }, // field is accessed from this document
+                    "input": { "$first": get_field(key.as_str()) }, // field is accessed from this document
                 },
             };
 
@@ -190,7 +191,7 @@ fn facet_pipelines_for_query(
                 value_expr
             };
 
-            (key.clone(), value_expr.into())
+            (key.to_string(), value_expr.into())
         })
         .collect();
 
@@ -235,11 +236,11 @@ fn pipeline_for_aggregate(
         Aggregate::ColumnCount { column, distinct } if distinct => Pipeline::from_iter(
             [
                 Some(Stage::Match(
-                    bson::doc! { &column: { "$exists": true, "$ne": null } },
+                    bson::doc! { column.as_str(): { "$exists": true, "$ne": null } },
                 )),
                 limit.map(Stage::Limit),
                 Some(Stage::Group {
-                    key_expression: field_ref(&column),
+                    key_expression: field_ref(column.as_str()),
                     accumulators: [].into(),
                 }),
                 Some(Stage::Count(RESULT_FIELD.to_string())),
@@ -251,7 +252,7 @@ fn pipeline_for_aggregate(
         Aggregate::ColumnCount { column, .. } => Pipeline::from_iter(
             [
                 Some(Stage::Match(
-                    bson::doc! { &column: { "$exists": true, "$ne": null } },
+                    bson::doc! { column.as_str(): { "$exists": true, "$ne": null } },
                 )),
                 limit.map(Stage::Limit),
                 Some(Stage::Count(RESULT_FIELD.to_string())),
@@ -266,11 +267,11 @@ fn pipeline_for_aggregate(
             use AggregationFunction::*;
 
             let accumulator = match function {
-                Avg => Accumulator::Avg(field_ref(&column)),
+                Avg => Accumulator::Avg(field_ref(column.as_str())),
                 Count => Accumulator::Count,
-                Min => Accumulator::Min(field_ref(&column)),
-                Max => Accumulator::Max(field_ref(&column)),
-                Sum => Accumulator::Sum(field_ref(&column)),
+                Min => Accumulator::Min(field_ref(column.as_str())),
+                Max => Accumulator::Max(field_ref(column.as_str())),
+                Sum => Accumulator::Sum(field_ref(column.as_str())),
             };
             Pipeline::from_iter(
                 [
diff --git a/crates/mongodb-agent-common/src/query/query_target.rs b/crates/mongodb-agent-common/src/query/query_target.rs
index ab4f53bc..b48fa7c3 100644
--- a/crates/mongodb-agent-common/src/query/query_target.rs
+++ b/crates/mongodb-agent-common/src/query/query_target.rs
@@ -7,11 +7,11 @@ use crate::mongo_query_plan::{MongoConfiguration, QueryPlan};
 
 #[derive(Clone, Debug)]
 pub enum QueryTarget<'a> {
-    Collection(String),
+    Collection(ndc_models::CollectionName),
     NativeQuery {
-        name: String,
+        name: ndc_models::CollectionName,
         native_query: &'a NativeQuery,
-        arguments: &'a BTreeMap<String, Argument>,
+        arguments: &'a BTreeMap<ndc_models::ArgumentName, Argument>,
     },
 }
 
@@ -31,12 +31,10 @@ impl QueryTarget<'_> {
         }
     }
 
-    pub fn input_collection(&self) -> Option<&str> {
+    pub fn input_collection(&self) -> Option<&ndc_models::CollectionName> {
         match self {
             QueryTarget::Collection(collection_name) => Some(collection_name),
-            QueryTarget::NativeQuery { native_query, .. } => {
-                native_query.input_collection.as_deref()
-            }
+            QueryTarget::NativeQuery { native_query, .. } => native_query.input_collection.as_ref(),
         }
     }
 }
diff --git a/crates/mongodb-agent-common/src/query/query_variable_name.rs b/crates/mongodb-agent-common/src/query/query_variable_name.rs
index 1778a700..bacaccbe 100644
--- a/crates/mongodb-agent-common/src/query/query_variable_name.rs
+++ b/crates/mongodb-agent-common/src/query/query_variable_name.rs
@@ -17,7 +17,7 @@ use crate::{
 /// - reproducibility: the same input name and type must always produce the same output name
 /// - distinct outputs: inputs with different types (or names) must produce different output names
 /// - It must produce a valid MongoDB variable name (see https://www.mongodb.com/docs/manual/reference/aggregation-variables/)
-pub fn query_variable_name(name: &str, variable_type: &Type) -> String {
+pub fn query_variable_name(name: &ndc_models::VariableName, variable_type: &Type) -> String {
     variable(&format!("{}_{}", name, type_name(variable_type)))
 }
 
@@ -52,8 +52,8 @@ mod tests {
     proptest! {
         #[test]
         fn variable_names_are_reproducible(variable_name: String, variable_type in arb_plan_type()) {
-            let a = query_variable_name(&variable_name, &variable_type);
-            let b = query_variable_name(&variable_name, &variable_type);
+            let a = query_variable_name(&variable_name.as_str().into(), &variable_type);
+            let b = query_variable_name(&variable_name.into(), &variable_type);
             prop_assert_eq!(a, b)
         }
     }
@@ -64,8 +64,8 @@ mod tests {
             (name_a, name_b) in (any::<String>(), any::<String>()).prop_filter("names are equale", |(a, b)| a != b),
             variable_type in arb_plan_type()
         ) {
-            let a = query_variable_name(&name_a, &variable_type);
-            let b = query_variable_name(&name_b, &variable_type);
+            let a = query_variable_name(&name_a.into(), &variable_type);
+            let b = query_variable_name(&name_b.into(), &variable_type);
             prop_assert_ne!(a, b)
         }
     }
@@ -76,8 +76,8 @@ mod tests {
             variable_name: String,
             (type_a, type_b) in (arb_plan_type(), arb_plan_type()).prop_filter("types are equal", |(a, b)| a != b)
         ) {
-            let a = query_variable_name(&variable_name, &type_a);
-            let b = query_variable_name(&variable_name, &type_b);
+            let a = query_variable_name(&variable_name.as_str().into(), &type_a);
+            let b = query_variable_name(&variable_name.into(), &type_b);
             prop_assert_ne!(a, b)
         }
     }
@@ -87,7 +87,7 @@ mod tests {
         fn variable_names_are_valid_for_mongodb_expressions(variable_name: String, variable_type in arb_plan_type()) {
             static VALID_NAME: Lazy<Regex> =
                 Lazy::new(|| Regex::new(r"^[a-z\P{ascii}][_a-zA-Z0-9\P{ascii}]*$").unwrap());
-            let name = query_variable_name(&variable_name, &variable_type);
+            let name = query_variable_name(&variable_name.into(), &variable_type);
             prop_assert!(VALID_NAME.is_match(&name))
         }
     }
diff --git a/crates/mongodb-agent-common/src/query/relations.rs b/crates/mongodb-agent-common/src/query/relations.rs
index bcbee0dc..0dbf9ae3 100644
--- a/crates/mongodb-agent-common/src/query/relations.rs
+++ b/crates/mongodb-agent-common/src/query/relations.rs
@@ -61,9 +61,9 @@ pub fn pipeline_for_relations(
 }
 
 fn make_lookup_stage(
-    from: String,
-    column_mapping: &BTreeMap<String, String>,
-    r#as: String,
+    from: ndc_models::CollectionName,
+    column_mapping: &BTreeMap<ndc_models::FieldName, ndc_models::FieldName>,
+    r#as: ndc_models::RelationshipName,
     lookup_pipeline: Pipeline,
     scope: Option<&Scope>,
 ) -> Result<Stage> {
@@ -87,17 +87,17 @@ fn make_lookup_stage(
 
 // TODO: MDB-160 Replace uses of [safe_name] with [ColumnRef].
 fn single_column_mapping_lookup(
-    from: String,
-    source_selector: &str,
-    target_selector: &str,
-    r#as: String,
+    from: ndc_models::CollectionName,
+    source_selector: &ndc_models::FieldName,
+    target_selector: &ndc_models::FieldName,
+    r#as: ndc_models::RelationshipName,
     lookup_pipeline: Pipeline,
     scope: Option<&Scope>,
 ) -> Result<Stage> {
     Ok(Stage::Lookup {
-        from: Some(from),
-        local_field: Some(safe_name(source_selector)?.into_owned()),
-        foreign_field: Some(safe_name(target_selector)?.into_owned()),
+        from: Some(from.to_string()),
+        local_field: Some(safe_name(source_selector.as_str())?.into_owned()),
+        foreign_field: Some(safe_name(target_selector.as_str())?.into_owned()),
         r#let: scope.map(|scope| {
             doc! {
                 name_from_scope(scope): "$$ROOT"
@@ -108,14 +108,14 @@ fn single_column_mapping_lookup(
         } else {
             Some(lookup_pipeline)
         },
-        r#as,
+        r#as: r#as.to_string(),
     })
 }
 
 fn multiple_column_mapping_lookup(
-    from: String,
-    column_mapping: &BTreeMap<String, String>,
-    r#as: String,
+    from: ndc_models::CollectionName,
+    column_mapping: &BTreeMap<ndc_models::FieldName, ndc_models::FieldName>,
+    r#as: ndc_models::RelationshipName,
     lookup_pipeline: Pipeline,
     scope: Option<&Scope>,
 ) -> Result<Stage> {
@@ -123,8 +123,11 @@ fn multiple_column_mapping_lookup(
         .keys()
         .map(|local_field| {
             Ok((
-                variable(local_field),
-                Bson::String(format!("${}", safe_name(local_field)?.into_owned())),
+                variable(local_field.as_str()),
+                Bson::String(format!(
+                    "${}",
+                    safe_name(local_field.as_str())?.into_owned()
+                )),
             ))
         })
         .collect::<Result<_>>()?;
@@ -136,15 +139,16 @@ fn multiple_column_mapping_lookup(
     // Creating an intermediate Vec and sorting it is done just to help with testing.
     // A stable order for matchers makes it easier to assert equality between actual
     // and expected pipelines.
-    let mut column_pairs: Vec<(&String, &String)> = column_mapping.iter().collect();
+    let mut column_pairs: Vec<(&ndc_models::FieldName, &ndc_models::FieldName)> =
+        column_mapping.iter().collect();
     column_pairs.sort();
 
     let matchers: Vec<Document> = column_pairs
         .into_iter()
         .map(|(local_field, remote_field)| {
             Ok(doc! { "$eq": [
-                format!("$${}", variable(local_field)),
-                format!("${}", safe_name(remote_field)?)
+                format!("$${}", variable(local_field.as_str())),
+                format!("${}", safe_name(remote_field.as_str())?)
             ] })
         })
         .collect::<Result<_>>()?;
@@ -162,12 +166,12 @@ fn multiple_column_mapping_lookup(
     let pipeline: Option<Pipeline> = pipeline.into();
 
     Ok(Stage::Lookup {
-        from: Some(from),
+        from: Some(from.to_string()),
         local_field: None,
         foreign_field: None,
         r#let: let_bindings.into(),
         pipeline,
-        r#as,
+        r#as: r#as.to_string(),
     })
 }
 
diff --git a/crates/mongodb-agent-common/src/query/response.rs b/crates/mongodb-agent-common/src/query/response.rs
index 92e143d4..dc386484 100644
--- a/crates/mongodb-agent-common/src/query/response.rs
+++ b/crates/mongodb-agent-common/src/query/response.rs
@@ -63,7 +63,7 @@ pub fn serialize_query_response(
                 let row_set = bson::from_document(document)?;
                 serialize_row_set_with_aggregates(
                     mode,
-                    &[collection_name],
+                    &[collection_name.as_str()],
                     &query_plan.query,
                     row_set,
                 )
@@ -135,16 +135,16 @@ fn serialize_row_set_with_aggregates(
 fn serialize_aggregates(
     mode: ExtendedJsonMode,
     path: &[&str],
-    _query_aggregates: &IndexMap<String, Aggregate>,
+    _query_aggregates: &IndexMap<ndc_models::FieldName, Aggregate>,
     value: Bson,
-) -> Result<IndexMap<String, serde_json::Value>> {
+) -> Result<IndexMap<ndc_models::FieldName, serde_json::Value>> {
     let aggregates_type = type_for_aggregates()?;
     let json = bson_to_json(mode, &aggregates_type, value)?;
 
     // The NDC type uses an IndexMap for aggregate values; we need to convert the map
     // underlying the Value::Object value to an IndexMap
     let aggregate_values = match json {
-        serde_json::Value::Object(obj) => obj.into_iter().collect(),
+        serde_json::Value::Object(obj) => obj.into_iter().map(|(k, v)| (k.into(), v)).collect(),
         _ => Err(QueryResponseError::AggregatesNotObject {
             path: path_to_owned(path),
         })?,
@@ -155,9 +155,9 @@ fn serialize_aggregates(
 fn serialize_rows(
     mode: ExtendedJsonMode,
     path: &[&str],
-    query_fields: &IndexMap<String, Field>,
+    query_fields: &IndexMap<ndc_models::FieldName, Field>,
     docs: Vec<bson::Document>,
-) -> Result<Vec<IndexMap<String, RowFieldValue>>> {
+) -> Result<Vec<IndexMap<ndc_models::FieldName, RowFieldValue>>> {
     let row_type = type_for_row(path, query_fields)?;
 
     docs.into_iter()
@@ -168,7 +168,7 @@ fn serialize_rows(
             let index_map = match json {
                 serde_json::Value::Object(obj) => obj
                     .into_iter()
-                    .map(|(key, value)| (key, RowFieldValue(value)))
+                    .map(|(key, value)| (key.into(), RowFieldValue(value)))
                     .collect(),
                 _ => unreachable!(),
             };
@@ -179,18 +179,18 @@ fn serialize_rows(
 
 fn type_for_row_set(
     path: &[&str],
-    aggregates: &Option<IndexMap<String, Aggregate>>,
-    fields: &Option<IndexMap<String, Field>>,
+    aggregates: &Option<IndexMap<ndc_models::FieldName, Aggregate>>,
+    fields: &Option<IndexMap<ndc_models::FieldName, Field>>,
 ) -> Result<Type> {
     let mut type_fields = BTreeMap::new();
 
     if aggregates.is_some() {
-        type_fields.insert("aggregates".to_owned(), type_for_aggregates()?);
+        type_fields.insert("aggregates".into(), type_for_aggregates()?);
     }
 
     if let Some(query_fields) = fields {
         let row_type = type_for_row(path, query_fields)?;
-        type_fields.insert("rows".to_owned(), Type::ArrayOf(Box::new(row_type)));
+        type_fields.insert("rows".into(), Type::ArrayOf(Box::new(row_type)));
     }
 
     Ok(Type::Object(ObjectType {
@@ -204,12 +204,15 @@ fn type_for_aggregates() -> Result<Type> {
     Ok(Type::Scalar(MongoScalarType::ExtendedJSON))
 }
 
-fn type_for_row(path: &[&str], query_fields: &IndexMap<String, Field>) -> Result<Type> {
+fn type_for_row(
+    path: &[&str],
+    query_fields: &IndexMap<ndc_models::FieldName, Field>,
+) -> Result<Type> {
     let fields = query_fields
         .iter()
         .map(|(field_name, field_definition)| {
             let field_type = type_for_field(
-                &append_to_path(path, [field_name.as_ref()]),
+                &append_to_path(path, [field_name.as_str()]),
                 field_definition,
             )?;
             Ok((field_name.clone(), field_type))
diff --git a/crates/mongodb-agent-common/src/query/serialization/bson_to_json.rs b/crates/mongodb-agent-common/src/query/serialization/bson_to_json.rs
index d1b4ebbc..ead29d93 100644
--- a/crates/mongodb-agent-common/src/query/serialization/bson_to_json.rs
+++ b/crates/mongodb-agent-common/src/query/serialization/bson_to_json.rs
@@ -129,7 +129,7 @@ fn convert_object(mode: ExtendedJsonMode, object_type: &ObjectType, value: Bson)
         })
         .map(|((field_name, field_type), field_value_result)| {
             Ok((
-                field_name.to_owned(),
+                field_name.to_string(),
                 bson_to_json(mode, field_type, field_value_result?)?,
             ))
         })
@@ -142,17 +142,17 @@ fn convert_object(mode: ExtendedJsonMode, object_type: &ObjectType, value: Bson)
 // nullable.
 fn get_object_field_value(
     object_type: &ObjectType,
-    (field_name, field_type): (&str, &Type),
+    (field_name, field_type): (&ndc_models::FieldName, &Type),
     doc: &bson::Document,
 ) -> Result<Option<Bson>> {
-    let value = doc.get(field_name);
+    let value = doc.get(field_name.as_str());
     if value.is_none() && is_nullable(field_type) {
         return Ok(None);
     }
     Ok(Some(value.cloned().ok_or_else(|| {
         BsonToJsonError::MissingObjectField(
             Type::Object(object_type.clone()),
-            field_name.to_owned(),
+            field_name.to_string(),
         )
     })?))
 }
@@ -233,7 +233,7 @@ mod tests {
         let expected_type = Type::Object(ObjectType {
             name: Some("test_object".into()),
             fields: [(
-                "field".to_owned(),
+                "field".into(),
                 Type::Nullable(Box::new(Type::Scalar(MongoScalarType::Bson(
                     BsonScalarType::String,
                 )))),
diff --git a/crates/mongodb-agent-common/src/query/serialization/json_to_bson.rs b/crates/mongodb-agent-common/src/query/serialization/json_to_bson.rs
index ac6dad86..05a75b5c 100644
--- a/crates/mongodb-agent-common/src/query/serialization/json_to_bson.rs
+++ b/crates/mongodb-agent-common/src/query/serialization/json_to_bson.rs
@@ -136,7 +136,7 @@ fn convert_object(object_type: &ObjectType, value: Value) -> Result<Bson> {
         })
         .map(|(name, field_type, field_value_result)| {
             Ok((
-                name.to_owned(),
+                name.to_string(),
                 json_to_bson(field_type, field_value_result?)?,
             ))
         })
@@ -149,18 +149,18 @@ fn convert_object(object_type: &ObjectType, value: Value) -> Result<Bson> {
 // nullable.
 fn get_object_field_value(
     object_type: &ObjectType,
-    field_name: &str,
+    field_name: &ndc_models::FieldName,
     field_type: &Type,
     object: &BTreeMap<String, Value>,
 ) -> Result<Option<Value>> {
-    let value = object.get(field_name);
+    let value = object.get(field_name.as_str());
     if value.is_none() && is_nullable(field_type) {
         return Ok(None);
     }
     Ok(Some(value.cloned().ok_or_else(|| {
         JsonToBsonError::MissingObjectField(
             Type::Object(object_type.clone()),
-            field_name.to_owned(),
+            field_name.to_string(),
         )
     })?))
 }
@@ -241,7 +241,7 @@ mod tests {
     #[allow(clippy::approx_constant)]
     fn deserializes_specialized_scalar_types() -> anyhow::Result<()> {
         let object_type = ObjectType {
-            name: Some("scalar_test".to_owned()),
+            name: Some("scalar_test".into()),
             fields: [
                 ("double", BsonScalarType::Double),
                 ("int", BsonScalarType::Int),
@@ -263,7 +263,7 @@ mod tests {
                 ("symbol", BsonScalarType::Symbol),
             ]
             .into_iter()
-            .map(|(name, t)| (name.to_owned(), Type::Scalar(MongoScalarType::Bson(t))))
+            .map(|(name, t)| (name.into(), Type::Scalar(MongoScalarType::Bson(t))))
             .collect(),
         };
 
@@ -369,9 +369,9 @@ mod tests {
     #[test]
     fn deserializes_object_with_missing_nullable_field() -> anyhow::Result<()> {
         let expected_type = Type::Object(ObjectType {
-            name: Some("test_object".to_owned()),
+            name: Some("test_object".into()),
             fields: [(
-                "field".to_owned(),
+                "field".into(),
                 Type::Nullable(Box::new(Type::Scalar(MongoScalarType::Bson(
                     BsonScalarType::String,
                 )))),
diff --git a/crates/mongodb-agent-common/src/scalar_types_capabilities.rs b/crates/mongodb-agent-common/src/scalar_types_capabilities.rs
index eaf41183..34b08b12 100644
--- a/crates/mongodb-agent-common/src/scalar_types_capabilities.rs
+++ b/crates/mongodb-agent-common/src/scalar_types_capabilities.rs
@@ -4,7 +4,8 @@ use itertools::Either;
 use lazy_static::lazy_static;
 use mongodb_support::BsonScalarType;
 use ndc_models::{
-    AggregateFunctionDefinition, ComparisonOperatorDefinition, ScalarType, Type, TypeRepresentation,
+    AggregateFunctionDefinition, AggregateFunctionName, ComparisonOperatorDefinition,
+    ComparisonOperatorName, ScalarType, Type, TypeRepresentation,
 };
 
 use crate::aggregation_function::{AggregationFunction, AggregationFunction as A};
@@ -13,19 +14,19 @@ use crate::comparison_function::{ComparisonFunction, ComparisonFunction as C};
 use BsonScalarType as S;
 
 lazy_static! {
-    pub static ref SCALAR_TYPES: BTreeMap<String, ScalarType> = scalar_types();
+    pub static ref SCALAR_TYPES: BTreeMap<ndc_models::ScalarTypeName, ScalarType> = scalar_types();
 }
 
-pub fn scalar_types() -> BTreeMap<String, ScalarType> {
+pub fn scalar_types() -> BTreeMap<ndc_models::ScalarTypeName, ScalarType> {
     enum_iterator::all::<BsonScalarType>()
         .map(make_scalar_type)
         .chain([extended_json_scalar_type()])
         .collect::<BTreeMap<_, _>>()
 }
 
-fn extended_json_scalar_type() -> (String, ScalarType) {
+fn extended_json_scalar_type() -> (ndc_models::ScalarTypeName, ScalarType) {
     (
-        mongodb_support::EXTENDED_JSON_TYPE_NAME.to_owned(),
+        mongodb_support::EXTENDED_JSON_TYPE_NAME.into(),
         ScalarType {
             representation: Some(TypeRepresentation::JSON),
             aggregate_functions: BTreeMap::new(),
@@ -34,14 +35,14 @@ fn extended_json_scalar_type() -> (String, ScalarType) {
     )
 }
 
-fn make_scalar_type(bson_scalar_type: BsonScalarType) -> (String, ScalarType) {
+fn make_scalar_type(bson_scalar_type: BsonScalarType) -> (ndc_models::ScalarTypeName, ScalarType) {
     let scalar_type_name = bson_scalar_type.graphql_name();
     let scalar_type = ScalarType {
         representation: bson_scalar_type_representation(bson_scalar_type),
         aggregate_functions: bson_aggregation_functions(bson_scalar_type),
         comparison_operators: bson_comparison_operators(bson_scalar_type),
     };
-    (scalar_type_name.to_owned(), scalar_type)
+    (scalar_type_name.into(), scalar_type)
 }
 
 fn bson_scalar_type_representation(bson_scalar_type: BsonScalarType) -> Option<TypeRepresentation> {
@@ -70,10 +71,10 @@ fn bson_scalar_type_representation(bson_scalar_type: BsonScalarType) -> Option<T
 
 fn bson_comparison_operators(
     bson_scalar_type: BsonScalarType,
-) -> BTreeMap<String, ComparisonOperatorDefinition> {
+) -> BTreeMap<ComparisonOperatorName, ComparisonOperatorDefinition> {
     comparison_operators(bson_scalar_type)
         .map(|(comparison_fn, arg_type)| {
-            let fn_name = comparison_fn.graphql_name().to_owned();
+            let fn_name = comparison_fn.graphql_name().into();
             match comparison_fn {
                 ComparisonFunction::Equal => (fn_name, ComparisonOperatorDefinition::Equal),
                 _ => (
@@ -89,20 +90,20 @@ fn bson_comparison_operators(
 
 fn bson_aggregation_functions(
     bson_scalar_type: BsonScalarType,
-) -> BTreeMap<String, AggregateFunctionDefinition> {
+) -> BTreeMap<AggregateFunctionName, AggregateFunctionDefinition> {
     aggregate_functions(bson_scalar_type)
         .map(|(fn_name, result_type)| {
             let aggregation_definition = AggregateFunctionDefinition {
                 result_type: bson_to_named_type(result_type),
             };
-            (fn_name.graphql_name().to_owned(), aggregation_definition)
+            (fn_name.graphql_name().into(), aggregation_definition)
         })
         .collect()
 }
 
 fn bson_to_named_type(bson_scalar_type: BsonScalarType) -> Type {
     Type::Named {
-        name: bson_scalar_type.graphql_name().to_owned(),
+        name: bson_scalar_type.graphql_name().into(),
     }
 }
 
diff --git a/crates/mongodb-agent-common/src/test_helpers.rs b/crates/mongodb-agent-common/src/test_helpers.rs
index d1058709..cc78a049 100644
--- a/crates/mongodb-agent-common/src/test_helpers.rs
+++ b/crates/mongodb-agent-common/src/test_helpers.rs
@@ -28,7 +28,7 @@ pub fn make_nested_schema() -> MongoConfiguration {
         functions: Default::default(),
         object_types: BTreeMap::from([
             (
-                "Author".to_owned(),
+                "Author".into(),
                 object_type([
                     ("name", schema::Type::Scalar(BsonScalarType::String)),
                     ("address", schema::Type::Object("Address".into())),
@@ -75,7 +75,7 @@ pub fn make_nested_schema() -> MongoConfiguration {
                 ]),
             ),
             (
-                "appearances".to_owned(),
+                "appearances".into(),
                 object_type([("authorId", schema::Type::Scalar(BsonScalarType::ObjectId))]),
             ),
         ]),
diff --git a/crates/mongodb-connector/src/capabilities.rs b/crates/mongodb-connector/src/capabilities.rs
index 1ee78543..460be3cd 100644
--- a/crates/mongodb-connector/src/capabilities.rs
+++ b/crates/mongodb-connector/src/capabilities.rs
@@ -1,30 +1,27 @@
 use ndc_sdk::models::{
-    Capabilities, CapabilitiesResponse, LeafCapability, NestedFieldCapabilities, QueryCapabilities,
+    Capabilities, LeafCapability, NestedFieldCapabilities, QueryCapabilities,
     RelationshipCapabilities,
 };
 
-pub fn mongo_capabilities_response() -> CapabilitiesResponse {
-    ndc_sdk::models::CapabilitiesResponse {
-        version: "0.1.4".to_owned(),
-        capabilities: Capabilities {
-            query: QueryCapabilities {
-                aggregates: Some(LeafCapability {}),
-                variables: Some(LeafCapability {}),
-                explain: Some(LeafCapability {}),
-                nested_fields: NestedFieldCapabilities {
-                    filter_by: Some(LeafCapability {}),
-                    order_by: Some(LeafCapability {}),
-                    aggregates: None,
-                },
+pub fn mongo_capabilities() -> Capabilities {
+    Capabilities {
+        query: QueryCapabilities {
+            aggregates: Some(LeafCapability {}),
+            variables: Some(LeafCapability {}),
+            explain: Some(LeafCapability {}),
+            nested_fields: NestedFieldCapabilities {
+                filter_by: Some(LeafCapability {}),
+                order_by: Some(LeafCapability {}),
+                aggregates: None,
             },
-            mutation: ndc_sdk::models::MutationCapabilities {
-                transactional: None,
-                explain: None,
-            },
-            relationships: Some(RelationshipCapabilities {
-                relation_comparisons: Some(LeafCapability {}),
-                order_by_aggregate: None,
-            }),
         },
+        mutation: ndc_sdk::models::MutationCapabilities {
+            transactional: None,
+            explain: None,
+        },
+        relationships: Some(RelationshipCapabilities {
+            relation_comparisons: Some(LeafCapability {}),
+            order_by_aggregate: None,
+        }),
     }
 }
diff --git a/crates/mongodb-connector/src/error_mapping.rs b/crates/mongodb-connector/src/error_mapping.rs
index 73bcd124..6db47afc 100644
--- a/crates/mongodb-connector/src/error_mapping.rs
+++ b/crates/mongodb-connector/src/error_mapping.rs
@@ -1,25 +1,43 @@
 use http::StatusCode;
-use mongodb_agent_common::interface_types::MongoAgentError;
-use ndc_sdk::connector::{ExplainError, QueryError};
+use mongodb_agent_common::interface_types::{ErrorResponse, MongoAgentError};
+use ndc_sdk::{
+    connector::{ExplainError, QueryError},
+    models,
+};
+use serde_json::Value;
 
 pub fn mongo_agent_error_to_query_error(error: MongoAgentError) -> QueryError {
     if let MongoAgentError::NotImplemented(e) = error {
-        return QueryError::UnsupportedOperation(e.to_owned());
+        return QueryError::UnsupportedOperation(error_response(e.to_owned()));
     }
     let (status, err) = error.status_and_error_response();
     match status {
-        StatusCode::BAD_REQUEST => QueryError::UnprocessableContent(err.message),
-        _ => QueryError::Other(Box::new(error)),
+        StatusCode::BAD_REQUEST => QueryError::UnprocessableContent(convert_error_response(err)),
+        _ => QueryError::Other(Box::new(error), Value::Object(Default::default())),
     }
 }
 
 pub fn mongo_agent_error_to_explain_error(error: MongoAgentError) -> ExplainError {
     if let MongoAgentError::NotImplemented(e) = error {
-        return ExplainError::UnsupportedOperation(e.to_owned());
+        return ExplainError::UnsupportedOperation(error_response(e.to_owned()));
     }
     let (status, err) = error.status_and_error_response();
     match status {
-        StatusCode::BAD_REQUEST => ExplainError::UnprocessableContent(err.message),
-        _ => ExplainError::Other(Box::new(error)),
+        StatusCode::BAD_REQUEST => ExplainError::UnprocessableContent(convert_error_response(err)),
+        _ => ExplainError::Other(Box::new(error), Value::Object(Default::default())),
+    }
+}
+
+pub fn error_response(message: String) -> models::ErrorResponse {
+    models::ErrorResponse {
+        message,
+        details: serde_json::Value::Object(Default::default()),
+    }
+}
+
+pub fn convert_error_response(err: ErrorResponse) -> models::ErrorResponse {
+    models::ErrorResponse {
+        message: err.message,
+        details: Value::Object(err.details.unwrap_or_default().into_iter().collect()),
     }
 }
diff --git a/crates/mongodb-connector/src/mongo_connector.rs b/crates/mongodb-connector/src/mongo_connector.rs
index 4c29c2cf..5df795a3 100644
--- a/crates/mongodb-connector/src/mongo_connector.rs
+++ b/crates/mongodb-connector/src/mongo_connector.rs
@@ -14,14 +14,17 @@ use ndc_sdk::{
     },
     json_response::JsonResponse,
     models::{
-        CapabilitiesResponse, ExplainResponse, MutationRequest, MutationResponse, QueryRequest,
+        Capabilities, ExplainResponse, MutationRequest, MutationResponse, QueryRequest,
         QueryResponse, SchemaResponse,
     },
 };
+use serde_json::Value;
 use tracing::instrument;
 
-use crate::error_mapping::{mongo_agent_error_to_explain_error, mongo_agent_error_to_query_error};
-use crate::{capabilities::mongo_capabilities_response, mutation::handle_mutation_request};
+use crate::error_mapping::{
+    error_response, mongo_agent_error_to_explain_error, mongo_agent_error_to_query_error,
+};
+use crate::{capabilities::mongo_capabilities, mutation::handle_mutation_request};
 
 #[derive(Clone, Default)]
 pub struct MongoConnector;
@@ -78,15 +81,18 @@ impl Connector for MongoConnector {
     ) -> Result<(), HealthError> {
         let status = check_health(state)
             .await
-            .map_err(|e| HealthError::Other(e.into()))?;
+            .map_err(|e| HealthError::Other(e.into(), Value::Object(Default::default())))?;
         match status.as_u16() {
             200..=299 => Ok(()),
-            s => Err(HealthError::Other(anyhow!("unhealthy status: {s}").into())),
+            s => Err(HealthError::Other(
+                anyhow!("unhealthy status: {s}").into(),
+                Value::Object(Default::default()),
+            )),
         }
     }
 
-    async fn get_capabilities() -> JsonResponse<CapabilitiesResponse> {
-        mongo_capabilities_response().into()
+    async fn get_capabilities() -> Capabilities {
+        mongo_capabilities()
     }
 
     #[instrument(err, skip_all)]
@@ -115,9 +121,9 @@ impl Connector for MongoConnector {
         _state: &Self::State,
         _request: MutationRequest,
     ) -> Result<JsonResponse<ExplainResponse>, ExplainError> {
-        Err(ExplainError::UnsupportedOperation(
+        Err(ExplainError::UnsupportedOperation(error_response(
             "Explain for mutations is not implemented yet".to_owned(),
-        ))
+        )))
     }
 
     #[instrument(err, skip_all)]
diff --git a/crates/mongodb-connector/src/mutation.rs b/crates/mongodb-connector/src/mutation.rs
index bc02348a..9f710812 100644
--- a/crates/mongodb-connector/src/mutation.rs
+++ b/crates/mongodb-connector/src/mutation.rs
@@ -20,6 +20,8 @@ use ndc_sdk::{
     },
 };
 
+use crate::error_mapping::error_response;
+
 pub async fn handle_mutation_request(
     config: &MongoConfiguration,
     state: &ConnectorState,
@@ -57,18 +59,22 @@ fn look_up_procedures<'a, 'b>(
                 fields,
             } => {
                 let native_mutation = config.native_mutations().get(name);
-                let procedure = native_mutation.ok_or(name).map(|native_mutation| {
-                    Procedure::from_native_mutation(native_mutation, arguments.clone())
-                })?;
+                let procedure = native_mutation
+                    .ok_or(name.to_string())
+                    .map(|native_mutation| {
+                        Procedure::from_native_mutation(native_mutation, arguments.clone())
+                    })?;
                 Ok((procedure, fields.as_ref()))
             }
         })
         .partition_result();
 
     if !not_found.is_empty() {
-        return Err(MutationError::UnprocessableContent(format!(
-            "request includes unknown mutations: {}",
-            not_found.join(", ")
+        return Err(MutationError::UnprocessableContent(error_response(
+            format!(
+                "request includes unknown mutations: {}",
+                not_found.join(", ")
+            ),
         )));
     }
 
@@ -85,7 +91,7 @@ async fn execute_procedure(
     let (result, result_type) = procedure
         .execute(database.clone())
         .await
-        .map_err(|err| MutationError::UnprocessableContent(err.to_string()))?;
+        .map_err(|err| MutationError::UnprocessableContent(error_response(err.to_string())))?;
 
     let rewritten_result = rewrite_response(requested_fields, result.into())?;
 
@@ -96,9 +102,9 @@ async fn execute_procedure(
             &result_type,
             fields.clone(),
         )
-        .map_err(|err| MutationError::UnprocessableContent(err.to_string()))?;
+        .map_err(|err| MutationError::UnprocessableContent(error_response(err.to_string())))?;
         type_for_nested_field(&[], &result_type, &plan_field)
-            .map_err(|err| MutationError::UnprocessableContent(err.to_string()))?
+            .map_err(|err| MutationError::UnprocessableContent(error_response(err.to_string())))?
     } else {
         result_type
     };
@@ -108,7 +114,7 @@ async fn execute_procedure(
         &requested_result_type,
         rewritten_result,
     )
-    .map_err(|err| MutationError::UnprocessableContent(err.to_string()))?;
+    .map_err(|err| MutationError::UnprocessableContent(error_response(err.to_string())))?;
 
     Ok(MutationOperationResults::Procedure {
         result: json_result,
@@ -132,10 +138,10 @@ fn rewrite_response(
         }
 
         (Some(NestedField::Object(_)), _) => Err(MutationError::UnprocessableContent(
-            "expected an object".to_owned(),
+            error_response("expected an object".to_owned()),
         )),
         (Some(NestedField::Array(_)), _) => Err(MutationError::UnprocessableContent(
-            "expected an array".to_owned(),
+            error_response("expected an array".to_owned()),
         )),
     }
 }
@@ -154,20 +160,20 @@ fn rewrite_doc(
                     fields,
                     arguments: _,
                 } => {
-                    let orig_value = doc.remove(column).ok_or_else(|| {
-                        MutationError::UnprocessableContent(format!(
+                    let orig_value = doc.remove(column.as_str()).ok_or_else(|| {
+                        MutationError::UnprocessableContent(error_response(format!(
                             "missing expected field from response: {name}"
-                        ))
+                        )))
                     })?;
                     rewrite_response(fields.as_ref(), orig_value)
                 }
                 ndc::Field::Relationship { .. } => Err(MutationError::UnsupportedOperation(
-                    "The MongoDB connector does not support relationship references in mutations"
-                        .to_owned(),
+                    error_response("The MongoDB connector does not support relationship references in mutations"
+                        .to_owned()),
                 )),
             }?;
 
-            Ok((name.clone(), field_value))
+            Ok((name.to_string(), field_value))
         })
         .try_collect()
 }
diff --git a/crates/ndc-query-plan/Cargo.toml b/crates/ndc-query-plan/Cargo.toml
index 06ec0331..7088e5ba 100644
--- a/crates/ndc-query-plan/Cargo.toml
+++ b/crates/ndc-query-plan/Cargo.toml
@@ -11,6 +11,7 @@ ndc-models = { workspace = true }
 nonempty = "^0.10"
 serde_json = "1"
 thiserror = "1"
+ref-cast = { workspace = true }
 
 [dev-dependencies]
 ndc-test-helpers = { path = "../ndc-test-helpers" }
diff --git a/crates/ndc-query-plan/src/plan_for_query_request/helpers.rs b/crates/ndc-query-plan/src/plan_for_query_request/helpers.rs
index f9c6d4b9..8dcf8edf 100644
--- a/crates/ndc-query-plan/src/plan_for_query_request/helpers.rs
+++ b/crates/ndc-query-plan/src/plan_for_query_request/helpers.rs
@@ -10,12 +10,12 @@ type Result<T> = std::result::Result<T, QueryPlanError>;
 
 pub fn find_object_field<'a, S>(
     object_type: &'a plan::ObjectType<S>,
-    field_name: &str,
+    field_name: &ndc::FieldName,
 ) -> Result<&'a plan::Type<S>> {
     object_type.fields.get(field_name).ok_or_else(|| {
         QueryPlanError::UnknownObjectTypeField {
             object_type: object_type.name.clone(),
-            field_name: field_name.to_string(),
+            field_name: field_name.clone(),
             path: Default::default(), // TODO: set a path for more helpful error reporting
         }
     })
@@ -23,8 +23,8 @@ pub fn find_object_field<'a, S>(
 
 pub fn find_object_field_path<'a, S>(
     object_type: &'a plan::ObjectType<S>,
-    field_name: &str,
-    field_path: &Option<Vec<String>>,
+    field_name: &ndc::FieldName,
+    field_path: &Option<Vec<ndc::FieldName>>,
 ) -> Result<&'a plan::Type<S>> {
     match field_path {
         None => find_object_field(object_type, field_name),
@@ -34,8 +34,8 @@ pub fn find_object_field_path<'a, S>(
 
 fn find_object_field_path_helper<'a, S>(
     object_type: &'a plan::ObjectType<S>,
-    field_name: &str,
-    field_path: &[String],
+    field_name: &ndc::FieldName,
+    field_path: &[ndc::FieldName],
 ) -> Result<&'a plan::Type<S>> {
     let field_type = find_object_field(object_type, field_name)?;
     match field_path {
@@ -49,8 +49,8 @@ fn find_object_field_path_helper<'a, S>(
 
 fn find_object_type<'a, S>(
     t: &'a plan::Type<S>,
-    parent_type: &Option<String>,
-    field_name: &str,
+    parent_type: &Option<ndc::ObjectTypeName>,
+    field_name: &ndc::FieldName,
 ) -> Result<&'a plan::ObjectType<S>> {
     match t {
         crate::Type::Scalar(_) => Err(QueryPlanError::ExpectedObjectTypeAtField {
@@ -69,8 +69,8 @@ fn find_object_type<'a, S>(
 }
 
 pub fn lookup_relationship<'a>(
-    relationships: &'a BTreeMap<String, ndc::Relationship>,
-    relationship: &str,
+    relationships: &'a BTreeMap<ndc::RelationshipName, ndc::Relationship>,
+    relationship: &ndc::RelationshipName,
 ) -> Result<&'a ndc::Relationship> {
     relationships
         .get(relationship)
diff --git a/crates/ndc-query-plan/src/plan_for_query_request/mod.rs b/crates/ndc-query-plan/src/plan_for_query_request/mod.rs
index f628123c..594cce4e 100644
--- a/crates/ndc-query-plan/src/plan_for_query_request/mod.rs
+++ b/crates/ndc-query-plan/src/plan_for_query_request/mod.rs
@@ -139,8 +139,8 @@ pub fn plan_for_query<T: QueryContext>(
 fn plan_for_aggregates<T: QueryContext>(
     context: &T,
     collection_object_type: &plan::ObjectType<T::ScalarType>,
-    ndc_aggregates: Option<IndexMap<String, ndc::Aggregate>>,
-) -> Result<Option<IndexMap<String, plan::Aggregate<T>>>> {
+    ndc_aggregates: Option<IndexMap<ndc::FieldName, ndc::Aggregate>>,
+) -> Result<Option<IndexMap<ndc::FieldName, plan::Aggregate<T>>>> {
     ndc_aggregates
         .map(|aggregates| -> Result<_> {
             aggregates
@@ -172,8 +172,7 @@ fn plan_for_aggregate<T: QueryContext>(
             function,
             field_path: _,
         } => {
-            let object_type_field_type =
-                find_object_field(collection_object_type, column.as_ref())?;
+            let object_type_field_type = find_object_field(collection_object_type, &column)?;
             // let column_scalar_type_name = get_scalar_type_name(&object_type_field.r#type)?;
             let (function, definition) =
                 context.find_aggregation_function_definition(object_type_field_type, &function)?;
@@ -191,9 +190,9 @@ fn plan_for_fields<T: QueryContext>(
     plan_state: &mut QueryPlanState<'_, T>,
     root_collection_object_type: &plan::ObjectType<T::ScalarType>,
     collection_object_type: &plan::ObjectType<T::ScalarType>,
-    ndc_fields: Option<IndexMap<String, ndc::Field>>,
-) -> Result<Option<IndexMap<String, plan::Field<T>>>> {
-    let plan_fields: Option<IndexMap<String, plan::Field<T>>> = ndc_fields
+    ndc_fields: Option<IndexMap<ndc::FieldName, ndc::Field>>,
+) -> Result<Option<IndexMap<ndc::FieldName, plan::Field<T>>>> {
+    let plan_fields: Option<IndexMap<ndc::FieldName, plan::Field<T>>> = ndc_fields
         .map(|fields| {
             fields
                 .into_iter()
@@ -308,8 +307,8 @@ fn plan_for_relationship_path<T: QueryContext>(
     root_collection_object_type: &plan::ObjectType<T::ScalarType>,
     object_type: &plan::ObjectType<T::ScalarType>,
     relationship_path: Vec<ndc::PathElement>,
-    requested_columns: Vec<String>, // columns to select from last path element
-) -> Result<(Vec<String>, ObjectType<T::ScalarType>)> {
+    requested_columns: Vec<ndc::FieldName>, // columns to select from last path element
+) -> Result<(Vec<ndc::RelationshipName>, ObjectType<T::ScalarType>)> {
     let end_of_relationship_path_object_type = relationship_path
         .last()
         .map(|last_path_element| {
@@ -345,8 +344,8 @@ fn plan_for_relationship_path_helper<T: QueryContext>(
     plan_state: &mut QueryPlanState<'_, T>,
     root_collection_object_type: &plan::ObjectType<T::ScalarType>,
     mut reversed_relationship_path: Vec<ndc::PathElement>,
-    requested_columns: Vec<String>, // columns to select from last path element
-) -> Result<VecDeque<String>> {
+    requested_columns: Vec<ndc::FieldName>, // columns to select from last path element
+) -> Result<VecDeque<ndc::RelationshipName>> {
     if reversed_relationship_path.is_empty() {
         return Ok(VecDeque::new());
     }
@@ -496,7 +495,7 @@ fn plan_for_binary_comparison<T: QueryContext>(
     root_collection_object_type: &plan::ObjectType<T::ScalarType>,
     object_type: &plan::ObjectType<T::ScalarType>,
     column: ndc::ComparisonTarget,
-    operator: String,
+    operator: ndc::ComparisonOperatorName,
     value: ndc::ComparisonValue,
 ) -> Result<plan::Expression<T>> {
     let comparison_target =
@@ -544,7 +543,8 @@ fn plan_for_comparison_target<T: QueryContext>(
                 path,
                 requested_columns,
             )?;
-            let field_type = find_object_field_path(&target_object_type, &name, &field_path)?.clone();
+            let field_type =
+                find_object_field_path(&target_object_type, &name, &field_path)?.clone();
             Ok(plan::ComparisonTarget::Column {
                 name,
                 field_path,
@@ -553,7 +553,8 @@ fn plan_for_comparison_target<T: QueryContext>(
             })
         }
         ndc::ComparisonTarget::RootCollectionColumn { name, field_path } => {
-            let field_type = find_object_field_path(root_collection_object_type, &name, &field_path)?.clone();
+            let field_type =
+                find_object_field_path(root_collection_object_type, &name, &field_path)?.clone();
             Ok(plan::ComparisonTarget::ColumnInScope {
                 name,
                 field_path,
@@ -630,7 +631,7 @@ fn plan_for_exists<T: QueryContext>(
                         (
                             comparison_target.column_name().to_owned(),
                             plan::Field::Column {
-                                column: comparison_target.column_name().to_string(),
+                                column: comparison_target.column_name().clone(),
                                 column_type: comparison_target.get_field_type().clone(),
                                 fields: None,
                             },
diff --git a/crates/ndc-query-plan/src/plan_for_query_request/plan_test_helpers/field.rs b/crates/ndc-query-plan/src/plan_for_query_request/plan_test_helpers/field.rs
index 46d1949a..3baaf035 100644
--- a/crates/ndc-query-plan/src/plan_for_query_request/plan_test_helpers/field.rs
+++ b/crates/ndc-query-plan/src/plan_for_query_request/plan_test_helpers/field.rs
@@ -4,7 +4,7 @@ macro_rules! field {
         (
             $name,
             $crate::Field::Column {
-                column: $name.to_owned(),
+                column: $name.into(),
                 column_type: $typ,
                 fields: None,
             },
@@ -14,7 +14,7 @@ macro_rules! field {
         (
             $name,
             $crate::Field::Column {
-                column: $column_name.to_owned(),
+                column: $column_name.into(),
                 column_type: $typ,
                 fields: None,
             },
@@ -24,7 +24,7 @@ macro_rules! field {
         (
             $name,
             $crate::Field::Column {
-                column: $column_name.to_owned(),
+                column: $column_name.into(),
                 column_type: $typ,
                 fields: Some($fields.into()),
             },
@@ -38,7 +38,7 @@ macro_rules! object {
         $crate::NestedField::Object($crate::NestedObject {
             fields: $fields
                 .into_iter()
-                .map(|(name, field)| (name.to_owned(), field))
+                .map(|(name, field)| (name.into(), field))
                 .collect(),
         })
     };
diff --git a/crates/ndc-query-plan/src/plan_for_query_request/plan_test_helpers/mod.rs b/crates/ndc-query-plan/src/plan_for_query_request/plan_test_helpers/mod.rs
index 31cee380..8518fd90 100644
--- a/crates/ndc-query-plan/src/plan_for_query_request/plan_test_helpers/mod.rs
+++ b/crates/ndc-query-plan/src/plan_for_query_request/plan_test_helpers/mod.rs
@@ -24,10 +24,10 @@ pub use self::{
 
 #[derive(Clone, Debug, Default)]
 pub struct TestContext {
-    pub collections: BTreeMap<String, ndc::CollectionInfo>,
-    pub functions: BTreeMap<String, (ndc::FunctionInfo, ndc::CollectionInfo)>,
-    pub procedures: BTreeMap<String, ndc::ProcedureInfo>,
-    pub object_types: BTreeMap<String, ndc::ObjectType>,
+    pub collections: BTreeMap<ndc::CollectionName, ndc::CollectionInfo>,
+    pub functions: BTreeMap<ndc::FunctionName, (ndc::FunctionInfo, ndc::CollectionInfo)>,
+    pub procedures: BTreeMap<ndc::ProcedureName, ndc::ProcedureInfo>,
+    pub object_types: BTreeMap<ndc::ObjectTypeName, ndc::ObjectType>,
 }
 
 impl ConnectorTypes for TestContext {
@@ -37,20 +37,21 @@ impl ConnectorTypes for TestContext {
 }
 
 impl QueryContext for TestContext {
-    fn lookup_scalar_type(type_name: &str) -> Option<Self::ScalarType> {
-        ScalarType::find_by_name(type_name)
+    fn lookup_scalar_type(type_name: &ndc::ScalarTypeName) -> Option<Self::ScalarType> {
+        ScalarType::find_by_name(type_name.as_str())
     }
 
     fn lookup_aggregation_function(
         &self,
         input_type: &Type<Self::ScalarType>,
-        function_name: &str,
+        function_name: &ndc::AggregateFunctionName,
     ) -> Result<(Self::AggregateFunction, &ndc::AggregateFunctionDefinition), QueryPlanError> {
-        let function = AggregateFunction::find_by_name(function_name).ok_or_else(|| {
-            QueryPlanError::UnknownAggregateFunction {
-                aggregate_function: function_name.to_owned(),
-            }
-        })?;
+        let function =
+            AggregateFunction::find_by_name(function_name.as_str()).ok_or_else(|| {
+                QueryPlanError::UnknownAggregateFunction {
+                    aggregate_function: function_name.to_owned(),
+                }
+            })?;
         let definition = scalar_type_name(input_type)
             .and_then(|name| SCALAR_TYPES.get(name))
             .and_then(|scalar_type_def| scalar_type_def.aggregate_functions.get(function_name))
@@ -63,12 +64,12 @@ impl QueryContext for TestContext {
     fn lookup_comparison_operator(
         &self,
         left_operand_type: &Type<Self::ScalarType>,
-        operator_name: &str,
+        operator_name: &ndc::ComparisonOperatorName,
     ) -> Result<(Self::ComparisonOperator, &ndc::ComparisonOperatorDefinition), QueryPlanError>
     where
         Self: Sized,
     {
-        let operator = ComparisonOperator::find_by_name(operator_name)
+        let operator = ComparisonOperator::find_by_name(operator_name.as_str())
             .ok_or_else(|| QueryPlanError::UnknownComparisonOperator(operator_name.to_owned()))?;
         let definition = scalar_type_name(left_operand_type)
             .and_then(|name| SCALAR_TYPES.get(name))
@@ -77,19 +78,19 @@ impl QueryContext for TestContext {
         Ok((operator, definition))
     }
 
-    fn collections(&self) -> &BTreeMap<String, ndc::CollectionInfo> {
+    fn collections(&self) -> &BTreeMap<ndc::CollectionName, ndc::CollectionInfo> {
         &self.collections
     }
 
-    fn functions(&self) -> &BTreeMap<String, (ndc::FunctionInfo, ndc::CollectionInfo)> {
+    fn functions(&self) -> &BTreeMap<ndc::FunctionName, (ndc::FunctionInfo, ndc::CollectionInfo)> {
         &self.functions
     }
 
-    fn object_types(&self) -> &BTreeMap<String, ndc::ObjectType> {
+    fn object_types(&self) -> &BTreeMap<ndc::ObjectTypeName, ndc::ObjectType> {
         &self.object_types
     }
 
-    fn procedures(&self) -> &BTreeMap<String, ndc::ProcedureInfo> {
+    fn procedures(&self) -> &BTreeMap<ndc::ProcedureName, ndc::ProcedureInfo> {
         &self.procedures
     }
 }
@@ -174,16 +175,16 @@ fn scalar_types() -> BTreeMap<String, ndc::ScalarType> {
             ndc::ScalarType {
                 representation: Some(TypeRepresentation::Float64),
                 aggregate_functions: [(
-                    AggregateFunction::Average.name().to_owned(),
+                    AggregateFunction::Average.name().into(),
                     ndc::AggregateFunctionDefinition {
                         result_type: ndc::Type::Named {
-                            name: ScalarType::Double.name().to_owned(),
+                            name: ScalarType::Double.name().into(),
                         },
                     },
                 )]
                 .into(),
                 comparison_operators: [(
-                    ComparisonOperator::Equal.name().to_owned(),
+                    ComparisonOperator::Equal.name().into(),
                     ndc::ComparisonOperatorDefinition::Equal,
                 )]
                 .into(),
@@ -194,16 +195,16 @@ fn scalar_types() -> BTreeMap<String, ndc::ScalarType> {
             ndc::ScalarType {
                 representation: Some(TypeRepresentation::Int32),
                 aggregate_functions: [(
-                    AggregateFunction::Average.name().to_owned(),
+                    AggregateFunction::Average.name().into(),
                     ndc::AggregateFunctionDefinition {
                         result_type: ndc::Type::Named {
-                            name: ScalarType::Double.name().to_owned(),
+                            name: ScalarType::Double.name().into(),
                         },
                     },
                 )]
                 .into(),
                 comparison_operators: [(
-                    ComparisonOperator::Equal.name().to_owned(),
+                    ComparisonOperator::Equal.name().into(),
                     ndc::ComparisonOperatorDefinition::Equal,
                 )]
                 .into(),
@@ -216,11 +217,11 @@ fn scalar_types() -> BTreeMap<String, ndc::ScalarType> {
                 aggregate_functions: Default::default(),
                 comparison_operators: [
                     (
-                        ComparisonOperator::Equal.name().to_owned(),
+                        ComparisonOperator::Equal.name().into(),
                         ndc::ComparisonOperatorDefinition::Equal,
                     ),
                     (
-                        ComparisonOperator::Regex.name().to_owned(),
+                        ComparisonOperator::Regex.name().into(),
                         ndc::ComparisonOperatorDefinition::Custom {
                             argument_type: named_type(ScalarType::String),
                         },
@@ -243,7 +244,7 @@ pub fn make_flat_schema() -> TestContext {
             (
                 "authors".into(),
                 ndc::CollectionInfo {
-                    name: "authors".to_owned(),
+                    name: "authors".into(),
                     description: None,
                     collection_type: "Author".into(),
                     arguments: Default::default(),
@@ -254,7 +255,7 @@ pub fn make_flat_schema() -> TestContext {
             (
                 "articles".into(),
                 ndc::CollectionInfo {
-                    name: "articles".to_owned(),
+                    name: "articles".into(),
                     description: None,
                     collection_type: "Article".into(),
                     arguments: Default::default(),
@@ -304,7 +305,7 @@ pub fn make_nested_schema() -> TestContext {
         functions: Default::default(),
         object_types: BTreeMap::from([
             (
-                "Author".to_owned(),
+                "Author".into(),
                 ndc_test_helpers::object_type([
                     ("name", named_type(ScalarType::String)),
                     ("address", named_type("Address")),
@@ -333,7 +334,7 @@ pub fn make_nested_schema() -> TestContext {
                 ]),
             ),
             (
-                "appearances".to_owned(),
+                "appearances".into(),
                 ndc_test_helpers::object_type([("authorId", named_type(ScalarType::Int))]),
             ),
         ]),
diff --git a/crates/ndc-query-plan/src/plan_for_query_request/plan_test_helpers/query.rs b/crates/ndc-query-plan/src/plan_for_query_request/plan_test_helpers/query.rs
index 4bad3cac..ddb9df8c 100644
--- a/crates/ndc-query-plan/src/plan_for_query_request/plan_test_helpers/query.rs
+++ b/crates/ndc-query-plan/src/plan_for_query_request/plan_test_helpers/query.rs
@@ -7,8 +7,8 @@ use crate::{
 
 #[derive(Clone, Debug, Default)]
 pub struct QueryBuilder<T: ConnectorTypes> {
-    aggregates: Option<IndexMap<String, Aggregate<T>>>,
-    fields: Option<IndexMap<String, Field<T>>>,
+    aggregates: Option<IndexMap<ndc_models::FieldName, Aggregate<T>>>,
+    fields: Option<IndexMap<ndc_models::FieldName, Field<T>>>,
     limit: Option<u32>,
     aggregates_limit: Option<u32>,
     offset: Option<u32>,
@@ -45,7 +45,7 @@ impl<T: ConnectorTypes> QueryBuilder<T> {
         self.fields = Some(
             fields
                 .into_iter()
-                .map(|(name, field)| (name.to_string(), field.into()))
+                .map(|(name, field)| (name.to_string().into(), field.into()))
                 .collect(),
         );
         self
@@ -55,7 +55,7 @@ impl<T: ConnectorTypes> QueryBuilder<T> {
         self.aggregates = Some(
             aggregates
                 .into_iter()
-                .map(|(name, aggregate)| (name.to_owned(), aggregate))
+                .map(|(name, aggregate)| (name.into(), aggregate))
                 .collect(),
         );
         self
diff --git a/crates/ndc-query-plan/src/plan_for_query_request/plan_test_helpers/relationships.rs b/crates/ndc-query-plan/src/plan_for_query_request/plan_test_helpers/relationships.rs
index b02263d0..2da3ff53 100644
--- a/crates/ndc-query-plan/src/plan_for_query_request/plan_test_helpers/relationships.rs
+++ b/crates/ndc-query-plan/src/plan_for_query_request/plan_test_helpers/relationships.rs
@@ -8,10 +8,10 @@ use super::QueryBuilder;
 
 #[derive(Clone, Debug)]
 pub struct RelationshipBuilder<T: ConnectorTypes> {
-    column_mapping: BTreeMap<String, String>,
+    column_mapping: BTreeMap<ndc_models::FieldName, ndc_models::FieldName>,
     relationship_type: RelationshipType,
-    target_collection: String,
-    arguments: BTreeMap<String, RelationshipArgument>,
+    target_collection: ndc_models::CollectionName,
+    arguments: BTreeMap<ndc_models::ArgumentName, RelationshipArgument>,
     query: QueryBuilder<T>,
 }
 
@@ -24,7 +24,7 @@ impl<T: ConnectorTypes> RelationshipBuilder<T> {
         RelationshipBuilder {
             column_mapping: Default::default(),
             relationship_type: RelationshipType::Array,
-            target_collection: target.to_owned(),
+            target_collection: target.into(),
             arguments: Default::default(),
             query: QueryBuilder::new(),
         }
@@ -46,7 +46,7 @@ impl<T: ConnectorTypes> RelationshipBuilder<T> {
     ) -> Self {
         self.column_mapping = column_mapping
             .into_iter()
-            .map(|(source, target)| (source.to_string(), target.to_string()))
+            .map(|(source, target)| (source.to_string().into(), target.to_string().into()))
             .collect();
         self
     }
@@ -61,7 +61,10 @@ impl<T: ConnectorTypes> RelationshipBuilder<T> {
         self
     }
 
-    pub fn arguments(mut self, arguments: BTreeMap<String, RelationshipArgument>) -> Self {
+    pub fn arguments(
+        mut self,
+        arguments: BTreeMap<ndc_models::ArgumentName, RelationshipArgument>,
+    ) -> Self {
         self.arguments = arguments;
         self
     }
diff --git a/crates/ndc-query-plan/src/plan_for_query_request/plan_test_helpers/type_helpers.rs b/crates/ndc-query-plan/src/plan_for_query_request/plan_test_helpers/type_helpers.rs
index 03be3369..7d0dc453 100644
--- a/crates/ndc-query-plan/src/plan_for_query_request/plan_test_helpers/type_helpers.rs
+++ b/crates/ndc-query-plan/src/plan_for_query_request/plan_test_helpers/type_helpers.rs
@@ -25,7 +25,7 @@ pub fn object_type(
         name: None,
         fields: fields
             .into_iter()
-            .map(|(name, field)| (name.to_string(), field.into()))
+            .map(|(name, field)| (name.to_string().into(), field.into()))
             .collect(),
     })
 }
diff --git a/crates/ndc-query-plan/src/plan_for_query_request/query_context.rs b/crates/ndc-query-plan/src/plan_for_query_request/query_context.rs
index 43336e85..b290e785 100644
--- a/crates/ndc-query-plan/src/plan_for_query_request/query_context.rs
+++ b/crates/ndc-query-plan/src/plan_for_query_request/query_context.rs
@@ -17,31 +17,31 @@ pub trait QueryContext: ConnectorTypes {
     /// Get the specific scalar type for this connector by name if the given name is a scalar type
     /// name. (This method will also be called for object type names in which case it should return
     /// `None`.)
-    fn lookup_scalar_type(type_name: &str) -> Option<Self::ScalarType>;
+    fn lookup_scalar_type(type_name: &ndc::ScalarTypeName) -> Option<Self::ScalarType>;
 
     fn lookup_aggregation_function(
         &self,
         input_type: &Type<Self::ScalarType>,
-        function_name: &str,
+        function_name: &ndc::AggregateFunctionName,
     ) -> Result<(Self::AggregateFunction, &ndc::AggregateFunctionDefinition)>;
 
     fn lookup_comparison_operator(
         &self,
         left_operand_type: &Type<Self::ScalarType>,
-        operator_name: &str,
+        operator_name: &ndc::ComparisonOperatorName,
     ) -> Result<(Self::ComparisonOperator, &ndc::ComparisonOperatorDefinition)>;
 
-    fn collections(&self) -> &BTreeMap<String, ndc::CollectionInfo>;
-    fn functions(&self) -> &BTreeMap<String, (ndc::FunctionInfo, ndc::CollectionInfo)>;
-    fn object_types(&self) -> &BTreeMap<String, ndc::ObjectType>;
-    fn procedures(&self) -> &BTreeMap<String, ndc::ProcedureInfo>;
+    fn collections(&self) -> &BTreeMap<ndc::CollectionName, ndc::CollectionInfo>;
+    fn functions(&self) -> &BTreeMap<ndc::FunctionName, (ndc::FunctionInfo, ndc::CollectionInfo)>;
+    fn object_types(&self) -> &BTreeMap<ndc::ObjectTypeName, ndc::ObjectType>;
+    fn procedures(&self) -> &BTreeMap<ndc::ProcedureName, ndc::ProcedureInfo>;
 
     /* Provided methods */
 
     fn find_aggregation_function_definition(
         &self,
         input_type: &Type<Self::ScalarType>,
-        function_name: &str,
+        function_name: &ndc::AggregateFunctionName,
     ) -> Result<(
         Self::AggregateFunction,
         plan::AggregateFunctionDefinition<Self>,
@@ -62,7 +62,7 @@ pub trait QueryContext: ConnectorTypes {
     fn find_comparison_operator(
         &self,
         left_operand_type: &Type<Self::ScalarType>,
-        op_name: &str,
+        op_name: &ndc::ComparisonOperatorName,
     ) -> Result<(
         Self::ComparisonOperator,
         plan::ComparisonOperatorDefinition<Self>,
@@ -84,7 +84,10 @@ pub trait QueryContext: ConnectorTypes {
         Ok((operator, plan_def))
     }
 
-    fn find_collection(&self, collection_name: &str) -> Result<&ndc::CollectionInfo> {
+    fn find_collection(
+        &self,
+        collection_name: &ndc::CollectionName,
+    ) -> Result<&ndc::CollectionInfo> {
         if let Some(collection) = self.collections().get(collection_name) {
             return Ok(collection);
         }
@@ -99,7 +102,7 @@ pub trait QueryContext: ConnectorTypes {
 
     fn find_collection_object_type(
         &self,
-        collection_name: &str,
+        collection_name: &ndc::CollectionName,
     ) -> Result<plan::ObjectType<Self::ScalarType>> {
         let collection = self.find_collection(collection_name)?;
         self.find_object_type(&collection.collection_type)
@@ -107,7 +110,7 @@ pub trait QueryContext: ConnectorTypes {
 
     fn find_object_type<'a>(
         &'a self,
-        object_type_name: &'a str,
+        object_type_name: &'a ndc::ObjectTypeName,
     ) -> Result<plan::ObjectType<Self::ScalarType>> {
         lookup_object_type(
             self.object_types(),
@@ -116,9 +119,9 @@ pub trait QueryContext: ConnectorTypes {
         )
     }
 
-    fn find_scalar_type(scalar_type_name: &str) -> Result<Self::ScalarType> {
+    fn find_scalar_type(scalar_type_name: &ndc::ScalarTypeName) -> Result<Self::ScalarType> {
         Self::lookup_scalar_type(scalar_type_name)
-            .ok_or_else(|| QueryPlanError::UnknownScalarType(scalar_type_name.to_owned()))
+            .ok_or_else(|| QueryPlanError::UnknownScalarType(scalar_type_name.clone()))
     }
 
     fn ndc_to_plan_type(&self, ndc_type: &ndc::Type) -> Result<plan::Type<Self::ScalarType>> {
diff --git a/crates/ndc-query-plan/src/plan_for_query_request/query_plan_error.rs b/crates/ndc-query-plan/src/plan_for_query_request/query_plan_error.rs
index f0107e00..d1f42a0c 100644
--- a/crates/ndc-query-plan/src/plan_for_query_request/query_plan_error.rs
+++ b/crates/ndc-query-plan/src/plan_for_query_request/query_plan_error.rs
@@ -1,3 +1,4 @@
+use ndc_models as ndc;
 use thiserror::Error;
 
 use super::unify_relationship_references::RelationshipUnificationError;
@@ -23,10 +24,10 @@ pub enum QueryPlanError {
     TypeMismatch(String),
 
     #[error("Unknown comparison operator, \"{0}\"")]
-    UnknownComparisonOperator(String),
+    UnknownComparisonOperator(ndc::ComparisonOperatorName),
 
     #[error("Unknown scalar type, \"{0}\"")]
-    UnknownScalarType(String),
+    UnknownScalarType(ndc::ScalarTypeName),
 
     #[error("Unknown object type, \"{0}\"")]
     UnknownObjectType(String),
@@ -37,8 +38,8 @@ pub enum QueryPlanError {
         at_path(path)
     )]
     UnknownObjectTypeField {
-        object_type: Option<String>,
-        field_name: String,
+        object_type: Option<ndc::ObjectTypeName>,
+        field_name: ndc::FieldName,
         path: Vec<String>,
     },
 
@@ -52,18 +53,20 @@ pub enum QueryPlanError {
     },
 
     #[error("Unknown aggregate function, \"{aggregate_function}\"")]
-    UnknownAggregateFunction { aggregate_function: String },
+    UnknownAggregateFunction {
+        aggregate_function: ndc::AggregateFunctionName,
+    },
 
     #[error("Query referenced a function, \"{0}\", but it has not been defined")]
-    UnspecifiedFunction(String),
+    UnspecifiedFunction(ndc::FunctionName),
 
     #[error("Query referenced a relationship, \"{0}\", but did not include relation metadata in `collection_relationships`")]
-    UnspecifiedRelation(String),
+    UnspecifiedRelation(ndc::RelationshipName),
 
-    #[error("Expected field {field_name} of object {} to be an object type. Got {got}.", parent_type.to_owned().unwrap_or("".to_owned()))]
+    #[error("Expected field {field_name} of object {} to be an object type. Got {got}.", parent_type.clone().map(|n| n.to_string()).unwrap_or("".to_owned()))]
     ExpectedObjectTypeAtField {
-        parent_type: Option<String>,
-        field_name: String,
+        parent_type: Option<ndc::ObjectTypeName>,
+        field_name: ndc::FieldName,
         got: String,
     },
 }
@@ -76,7 +79,7 @@ fn at_path(path: &[String]) -> String {
     }
 }
 
-fn in_object_type(type_name: Option<&String>) -> String {
+fn in_object_type(type_name: Option<&ndc::ObjectTypeName>) -> String {
     match type_name {
         Some(name) => format!(" in object type \"{name}\""),
         None => "".to_owned(),
diff --git a/crates/ndc-query-plan/src/plan_for_query_request/query_plan_state.rs b/crates/ndc-query-plan/src/plan_for_query_request/query_plan_state.rs
index e5a4c78c..a000fdc9 100644
--- a/crates/ndc-query-plan/src/plan_for_query_request/query_plan_state.rs
+++ b/crates/ndc-query-plan/src/plan_for_query_request/query_plan_state.rs
@@ -27,9 +27,9 @@ type Result<T> = std::result::Result<T, QueryPlanError>;
 #[derive(Debug)]
 pub struct QueryPlanState<'a, T: QueryContext> {
     pub context: &'a T,
-    pub collection_relationships: &'a BTreeMap<String, ndc::Relationship>,
+    pub collection_relationships: &'a BTreeMap<ndc::RelationshipName, ndc::Relationship>,
     pub scope: Scope,
-    relationships: BTreeMap<String, Relationship<T>>,
+    relationships: BTreeMap<ndc_models::RelationshipName, Relationship<T>>,
     unrelated_joins: Rc<RefCell<BTreeMap<String, UnrelatedJoin<T>>>>,
     relationship_name_counter: Rc<Cell<i32>>,
     scope_name_counter: Rc<Cell<i32>>,
@@ -39,7 +39,7 @@ pub struct QueryPlanState<'a, T: QueryContext> {
 impl<T: QueryContext> QueryPlanState<'_, T> {
     pub fn new<'a>(
         query_context: &'a T,
-        collection_relationships: &'a BTreeMap<String, ndc::Relationship>,
+        collection_relationships: &'a BTreeMap<ndc::RelationshipName, ndc::Relationship>,
     ) -> QueryPlanState<'a, T> {
         QueryPlanState {
             context: query_context,
@@ -78,10 +78,10 @@ impl<T: QueryContext> QueryPlanState<'_, T> {
     /// plan, and get back an identifier than can be used to access the joined collection.
     pub fn register_relationship(
         &mut self,
-        ndc_relationship_name: String,
-        arguments: BTreeMap<String, RelationshipArgument>,
+        ndc_relationship_name: ndc::RelationshipName,
+        arguments: BTreeMap<ndc::ArgumentName, RelationshipArgument>,
         query: Query<T>,
-    ) -> Result<String> {
+    ) -> Result<ndc::RelationshipName> {
         let ndc_relationship =
             lookup_relationship(self.collection_relationships, &ndc_relationship_name)?;
 
@@ -113,7 +113,7 @@ impl<T: QueryContext> QueryPlanState<'_, T> {
                         // relationship that we just removed.
                         self.relationships
                             .insert(existing_key, already_registered_relationship);
-                        let key = self.unique_relationship_name(ndc_relationship_name);
+                        let key = self.unique_relationship_name(ndc_relationship_name).into();
                         (key, relationship)
                     }
                 }
@@ -130,8 +130,8 @@ impl<T: QueryContext> QueryPlanState<'_, T> {
     /// plan, and get back an identifier than can be used to access the joined collection.
     pub fn register_unrelated_join(
         &mut self,
-        target_collection: String,
-        arguments: BTreeMap<String, RelationshipArgument>,
+        target_collection: ndc::CollectionName,
+        arguments: BTreeMap<ndc::ArgumentName, RelationshipArgument>,
         query: Query<T>,
     ) -> String {
         let join = UnrelatedJoin {
@@ -156,25 +156,25 @@ impl<T: QueryContext> QueryPlanState<'_, T> {
     /// a [crate::QueryPlan] so we can capture types for each variable.
     pub fn register_variable_use(
         &mut self,
-        variable_name: &str,
+        variable_name: &ndc::VariableName,
         expected_type: Type<T::ScalarType>,
     ) {
         self.register_variable_use_helper(variable_name, Some(expected_type))
     }
 
-    pub fn register_variable_use_of_unknown_type(&mut self, variable_name: &str) {
+    pub fn register_variable_use_of_unknown_type(&mut self, variable_name: &ndc::VariableName) {
         self.register_variable_use_helper(variable_name, None)
     }
 
     fn register_variable_use_helper(
         &mut self,
-        variable_name: &str,
+        variable_name: &ndc::VariableName,
         expected_type: Option<Type<T::ScalarType>>,
     ) {
         let mut type_map = self.variable_types.borrow_mut();
         match type_map.get_mut(variable_name) {
             None => {
-                type_map.insert(variable_name.to_string(), VecSet::singleton(expected_type));
+                type_map.insert(variable_name.clone(), VecSet::singleton(expected_type));
             }
             Some(entry) => {
                 entry.insert(expected_type);
@@ -183,7 +183,7 @@ impl<T: QueryContext> QueryPlanState<'_, T> {
     }
 
     /// Use this for subquery plans to get the relationships for each sub-query
-    pub fn into_relationships(self) -> BTreeMap<String, Relationship<T>> {
+    pub fn into_relationships(self) -> BTreeMap<ndc::RelationshipName, Relationship<T>> {
         self.relationships
     }
 
diff --git a/crates/ndc-query-plan/src/plan_for_query_request/tests.rs b/crates/ndc-query-plan/src/plan_for_query_request/tests.rs
index 82472f1b..1d5d1c6e 100644
--- a/crates/ndc-query-plan/src/plan_for_query_request/tests.rs
+++ b/crates/ndc-query-plan/src/plan_for_query_request/tests.rs
@@ -54,27 +54,27 @@ fn translates_query_request_relationships() -> Result<(), anyhow::Error> {
                 .order_by(vec![ndc::OrderByElement {
                     order_direction: OrderDirection::Asc,
                     target: OrderByTarget::Column {
-                        name: "advisor_name".to_owned(),
+                        name: "advisor_name".into(),
                         field_path: None,
                         path: vec![
-                            path_element("school_classes")
+                            path_element("school_classes".into())
                                 .predicate(binop(
                                     "Equal",
                                     target!(
                                         "_id",
                                         relations: [
                                         // path_element("school_classes"),
-                                        path_element("class_department"),
+                                        path_element("class_department".into()),
                                     ],
                                     ),
                                     column_value!(
                                         "math_department_id",
-                                        relations: [path_element("school_directory")],
+                                        relations: [path_element("school_directory".into())],
                                     ),
                                 ))
                                 .into(),
-                            path_element("class_students").into(),
-                            path_element("student_advisor").into(),
+                            path_element("class_students".into()).into(),
+                            path_element("student_advisor".into()).into(),
                         ],
                     },
                 }])
@@ -87,7 +87,7 @@ fn translates_query_request_relationships() -> Result<(), anyhow::Error> {
         .into();
 
     let expected = QueryPlan {
-        collection: "schools".to_owned(),
+        collection: "schools".into(),
         arguments: Default::default(),
         variables: None,
         variable_types: Default::default(),
@@ -119,11 +119,11 @@ fn translates_query_request_relationships() -> Result<(), anyhow::Error> {
             }),
             relationships: [
                 (
-                    "school_classes_0".to_owned(),
+                    "school_classes_0".into(),
                     Relationship {
-                        column_mapping: [("_id".to_owned(), "school_id".to_owned())].into(),
+                        column_mapping: [("_id".into(), "school_id".into())].into(),
                         relationship_type: RelationshipType::Array,
-                        target_collection: "classes".to_owned(),
+                        target_collection: "classes".into(),
                         arguments: Default::default(),
                         query: Query {
                             predicate: Some(plan::Expression::BinaryComparisonOperator {
@@ -202,10 +202,10 @@ fn translates_query_request_relationships() -> Result<(), anyhow::Error> {
                                     },
                                 ),
                                 (
-                                    "school_directory".to_owned(),
+                                    "school_directory".into(),
                                     Relationship {
-                                        target_collection: "directory".to_owned(),
-                                        column_mapping: [("_id".to_owned(), "school_id".to_owned())].into(),
+                                        target_collection: "directory".into(),
+                                        column_mapping: [("_id".into(), "school_id".into())].into(),
                                         relationship_type: RelationshipType::Object,
                                         arguments: Default::default(),
                                         query: Query {
@@ -223,11 +223,11 @@ fn translates_query_request_relationships() -> Result<(), anyhow::Error> {
                     },
                 ),
                 (
-                    "school_classes".to_owned(),
+                    "school_classes".into(),
                     Relationship {
-                        column_mapping: [("_id".to_owned(), "school_id".to_owned())].into(),
+                        column_mapping: [("_id".into(), "school_id".into())].into(),
                         relationship_type: RelationshipType::Array,
-                        target_collection: "classes".to_owned(),
+                        target_collection: "classes".into(),
                         arguments: Default::default(),
                         query: Query {
                             fields: Some(
@@ -260,11 +260,11 @@ fn translates_query_request_relationships() -> Result<(), anyhow::Error> {
                     },
                 ),
                 (
-                    "existence_check".to_owned(),
+                    "existence_check".into(),
                     Relationship {
-                        column_mapping: [("some_id".to_owned(), "_id".to_owned())].into(),
+                        column_mapping: [("some_id".into(), "_id".into())].into(),
                         relationship_type: RelationshipType::Array,
-                        target_collection: "some_collection".to_owned(),
+                        target_collection: "some_collection".into(),
                         arguments: Default::default(),
                         query: Query {
                             predicate: None,
@@ -312,12 +312,9 @@ fn translates_query_request_relationships() -> Result<(), anyhow::Error> {
         ]
         .into(),
         object_types: [
+            ("schools".into(), object_type([("_id", named_type("Int"))])),
             (
-                "schools".to_owned(),
-                object_type([("_id", named_type("Int"))]),
-            ),
-            (
-                "classes".to_owned(),
+                "classes".into(),
                 object_type([
                     ("_id", named_type("Int")),
                     ("school_id", named_type("Int")),
@@ -325,7 +322,7 @@ fn translates_query_request_relationships() -> Result<(), anyhow::Error> {
                 ]),
             ),
             (
-                "students".to_owned(),
+                "students".into(),
                 object_type([
                     ("_id", named_type("Int")),
                     ("class_id", named_type("Int")),
@@ -334,11 +331,11 @@ fn translates_query_request_relationships() -> Result<(), anyhow::Error> {
                 ]),
             ),
             (
-                "departments".to_owned(),
+                "departments".into(),
                 object_type([("_id", named_type("Int"))]),
             ),
             (
-                "directory".to_owned(),
+                "directory".into(),
                 object_type([
                     ("_id", named_type("Int")),
                     ("school_id", named_type("Int")),
@@ -346,14 +343,14 @@ fn translates_query_request_relationships() -> Result<(), anyhow::Error> {
                 ]),
             ),
             (
-                "advisors".to_owned(),
+                "advisors".into(),
                 object_type([
                     ("_id", named_type("Int")),
                     ("advisor_name", named_type("String")),
                 ]),
             ),
             (
-                "some_collection".to_owned(),
+                "some_collection".into(),
                 object_type([("_id", named_type("Int")), ("some_id", named_type("Int"))]),
             ),
         ]
@@ -580,7 +577,7 @@ fn translates_relationships_in_fields_predicates_and_orderings() -> Result<(), a
                         target: OrderByTarget::SingleColumnAggregate {
                             column: "year".into(),
                             function: "Average".into(),
-                            path: vec![path_element("author_articles").into()],
+                            path: vec![path_element("author_articles".into()).into()],
                             field_path: None,
                         },
                     },
@@ -765,7 +762,7 @@ fn translates_nested_fields() -> Result<(), anyhow::Error> {
                         plan::Field::Column {
                             column: "address".into(),
                             column_type: plan::Type::Object(
-                                query_context.find_object_type("Address")?,
+                                query_context.find_object_type(&"Address".into())?,
                             ),
                             fields: Some(plan::NestedField::Object(plan::NestedObject {
                                 fields: [(
@@ -787,7 +784,7 @@ fn translates_nested_fields() -> Result<(), anyhow::Error> {
                         plan::Field::Column {
                             column: "articles".into(),
                             column_type: plan::Type::ArrayOf(Box::new(plan::Type::Object(
-                                query_context.find_object_type("Article")?,
+                                query_context.find_object_type(&"Article".into())?,
                             ))),
                             fields: Some(plan::NestedField::Array(plan::NestedArray {
                                 fields: Box::new(plan::NestedField::Object(plan::NestedObject {
@@ -831,7 +828,7 @@ fn translates_nested_fields() -> Result<(), anyhow::Error> {
                             })),
                             column_type: plan::Type::ArrayOf(Box::new(plan::Type::ArrayOf(
                                 Box::new(plan::Type::Object(
-                                    query_context.find_object_type("Article")?,
+                                    query_context.find_object_type(&"Article".into())?,
                                 )),
                             ))),
                         },
@@ -864,7 +861,7 @@ fn translates_predicate_referencing_field_of_related_collection() -> anyhow::Res
                     field!("name"),
                 ]))])
                 .predicate(not(is_null(
-                    target!("name", relations: [path_element("author")]),
+                    target!("name", relations: [path_element("author".into())]),
                 ))),
         )
         .into();
diff --git a/crates/ndc-query-plan/src/plan_for_query_request/type_annotated_field.rs b/crates/ndc-query-plan/src/plan_for_query_request/type_annotated_field.rs
index 61589ef2..fa6de979 100644
--- a/crates/ndc-query-plan/src/plan_for_query_request/type_annotated_field.rs
+++ b/crates/ndc-query-plan/src/plan_for_query_request/type_annotated_field.rs
@@ -105,7 +105,7 @@ fn type_annotated_field_helper<T: QueryContext>(
 /// Translates [ndc::NestedField] to [Field]. The latter includes type annotations.
 pub fn type_annotated_nested_field<T: QueryContext>(
     query_context: &T,
-    collection_relationships: &BTreeMap<String, ndc::Relationship>,
+    collection_relationships: &BTreeMap<ndc::RelationshipName, ndc::Relationship>,
     result_type: &Type<T::ScalarType>,
     requested_fields: ndc::NestedField,
 ) -> Result<NestedField<T>> {
@@ -144,7 +144,7 @@ fn type_annotated_nested_field_helper<T: QueryContext>(
                                 root_collection_object_type,
                                 object_type,
                                 field.clone(),
-                                &append_to_path(path, [name.as_ref()]),
+                                &append_to_path(path, [name.to_string().as_ref()]),
                             )?,
                         )) as Result<_>
                     })
diff --git a/crates/ndc-query-plan/src/plan_for_query_request/unify_relationship_references.rs b/crates/ndc-query-plan/src/plan_for_query_request/unify_relationship_references.rs
index b011b2ba..e83010a8 100644
--- a/crates/ndc-query-plan/src/plan_for_query_request/unify_relationship_references.rs
+++ b/crates/ndc-query-plan/src/plan_for_query_request/unify_relationship_references.rs
@@ -15,25 +15,25 @@ use crate::{
 pub enum RelationshipUnificationError {
     #[error("relationship arguments mismatch")]
     ArgumentsMismatch {
-        a: BTreeMap<String, RelationshipArgument>,
-        b: BTreeMap<String, RelationshipArgument>,
+        a: BTreeMap<ndc_models::ArgumentName, RelationshipArgument>,
+        b: BTreeMap<ndc_models::ArgumentName, RelationshipArgument>,
     },
 
     #[error("relationships select fields with the same name, {field_name}, but that have different types")]
-    FieldTypeMismatch { field_name: String },
+    FieldTypeMismatch { field_name: ndc_models::FieldName },
 
     #[error("relationships select columns {column_a} and {column_b} with the same field name, {field_name}")]
     FieldColumnMismatch {
-        field_name: String,
-        column_a: String,
-        column_b: String,
+        field_name: ndc_models::FieldName,
+        column_a: ndc_models::FieldName,
+        column_b: ndc_models::FieldName,
     },
 
     #[error("relationship references have incompatible configurations: {}", .0.join(", "))]
     Mismatch(Vec<&'static str>),
 
     #[error("relationship references referenced different nested relationships with the same field name, {field_name}")]
-    RelationshipMismatch { field_name: String },
+    RelationshipMismatch { field_name: ndc_models::FieldName },
 }
 
 type Result<T> = std::result::Result<T, RelationshipUnificationError>;
@@ -65,9 +65,9 @@ where
 // being pessimistic, and if we get an error here we record the two relationships under separate
 // keys instead of recording one, unified relationship.
 fn unify_arguments(
-    a: BTreeMap<String, RelationshipArgument>,
-    b: BTreeMap<String, RelationshipArgument>,
-) -> Result<BTreeMap<String, RelationshipArgument>> {
+    a: BTreeMap<ndc_models::ArgumentName, RelationshipArgument>,
+    b: BTreeMap<ndc_models::ArgumentName, RelationshipArgument>,
+) -> Result<BTreeMap<ndc_models::ArgumentName, RelationshipArgument>> {
     if a != b {
         Err(RelationshipUnificationError::ArgumentsMismatch { a, b })
     } else {
@@ -120,9 +120,9 @@ where
 }
 
 fn unify_aggregates<T>(
-    a: Option<IndexMap<String, Aggregate<T>>>,
-    b: Option<IndexMap<String, Aggregate<T>>>,
-) -> Result<Option<IndexMap<String, Aggregate<T>>>>
+    a: Option<IndexMap<ndc_models::FieldName, Aggregate<T>>>,
+    b: Option<IndexMap<ndc_models::FieldName, Aggregate<T>>>,
+) -> Result<Option<IndexMap<ndc_models::FieldName, Aggregate<T>>>>
 where
     T: ConnectorTypes,
 {
@@ -134,9 +134,9 @@ where
 }
 
 fn unify_fields<T>(
-    a: Option<IndexMap<String, Field<T>>>,
-    b: Option<IndexMap<String, Field<T>>>,
-) -> Result<Option<IndexMap<String, Field<T>>>>
+    a: Option<IndexMap<ndc_models::FieldName, Field<T>>>,
+    b: Option<IndexMap<ndc_models::FieldName, Field<T>>>,
+) -> Result<Option<IndexMap<ndc_models::FieldName, Field<T>>>>
 where
     T: ConnectorTypes,
 {
@@ -144,9 +144,9 @@ where
 }
 
 fn unify_fields_some<T>(
-    fields_a: IndexMap<String, Field<T>>,
-    fields_b: IndexMap<String, Field<T>>,
-) -> Result<IndexMap<String, Field<T>>>
+    fields_a: IndexMap<ndc_models::FieldName, Field<T>>,
+    fields_b: IndexMap<ndc_models::FieldName, Field<T>>,
+) -> Result<IndexMap<ndc_models::FieldName, Field<T>>>
 where
     T: ConnectorTypes,
 {
@@ -163,7 +163,7 @@ where
     Ok(fields)
 }
 
-fn unify_field<T>(field_name: &str, a: Field<T>, b: Field<T>) -> Result<Field<T>>
+fn unify_field<T>(field_name: &ndc_models::FieldName, a: Field<T>, b: Field<T>) -> Result<Field<T>>
 where
     T: ConnectorTypes,
 {
diff --git a/crates/ndc-query-plan/src/query_plan.rs b/crates/ndc-query-plan/src/query_plan.rs
index 49200ff6..f200c754 100644
--- a/crates/ndc-query-plan/src/query_plan.rs
+++ b/crates/ndc-query-plan/src/query_plan.rs
@@ -22,9 +22,9 @@ pub trait ConnectorTypes {
     PartialEq(bound = "T::ScalarType: PartialEq")
 )]
 pub struct QueryPlan<T: ConnectorTypes> {
-    pub collection: String,
+    pub collection: ndc_models::CollectionName,
     pub query: Query<T>,
-    pub arguments: BTreeMap<String, Argument>,
+    pub arguments: BTreeMap<ndc_models::ArgumentName, Argument>,
     pub variables: Option<Vec<VariableSet>>,
 
     /// Types for values from the `variables` map as inferred by usages in the query request. It is
@@ -44,9 +44,9 @@ impl<T: ConnectorTypes> QueryPlan<T> {
     }
 }
 
-pub type Relationships<T> = BTreeMap<String, Relationship<T>>;
-pub type VariableSet = BTreeMap<String, serde_json::Value>;
-pub type VariableTypes<T> = BTreeMap<String, VecSet<Option<Type<T>>>>;
+pub type Relationships<T> = BTreeMap<ndc_models::RelationshipName, Relationship<T>>;
+pub type VariableSet = BTreeMap<ndc_models::VariableName, serde_json::Value>;
+pub type VariableTypes<T> = BTreeMap<ndc_models::VariableName, VecSet<Option<Type<T>>>>;
 
 #[derive(Derivative)]
 #[derivative(
@@ -56,8 +56,8 @@ pub type VariableTypes<T> = BTreeMap<String, VecSet<Option<Type<T>>>>;
     PartialEq(bound = "")
 )]
 pub struct Query<T: ConnectorTypes> {
-    pub aggregates: Option<IndexMap<String, Aggregate<T>>>,
-    pub fields: Option<IndexMap<String, Field<T>>>,
+    pub aggregates: Option<IndexMap<ndc_models::FieldName, Aggregate<T>>>,
+    pub fields: Option<IndexMap<ndc_models::FieldName, Field<T>>>,
     pub limit: Option<u32>,
     pub aggregates_limit: Option<u32>,
     pub offset: Option<u32>,
@@ -95,18 +95,18 @@ impl<T: ConnectorTypes> Query<T> {
 #[derive(Derivative)]
 #[derivative(Clone(bound = ""), Debug(bound = ""), PartialEq(bound = ""))]
 pub struct Relationship<T: ConnectorTypes> {
-    pub column_mapping: BTreeMap<String, String>,
+    pub column_mapping: BTreeMap<ndc_models::FieldName, ndc_models::FieldName>,
     pub relationship_type: RelationshipType,
-    pub target_collection: String,
-    pub arguments: BTreeMap<String, RelationshipArgument>,
+    pub target_collection: ndc_models::CollectionName,
+    pub arguments: BTreeMap<ndc_models::ArgumentName, RelationshipArgument>,
     pub query: Query<T>,
 }
 
 #[derive(Derivative)]
 #[derivative(Clone(bound = ""), Debug(bound = ""), PartialEq(bound = ""))]
 pub struct UnrelatedJoin<T: ConnectorTypes> {
-    pub target_collection: String,
-    pub arguments: BTreeMap<String, RelationshipArgument>,
+    pub target_collection: ndc_models::CollectionName,
+    pub arguments: BTreeMap<ndc_models::ArgumentName, RelationshipArgument>,
     pub query: Query<T>,
 }
 
@@ -121,13 +121,13 @@ pub enum Scope {
 pub enum Aggregate<T: ConnectorTypes> {
     ColumnCount {
         /// The column to apply the count aggregate function to
-        column: String,
+        column: ndc_models::FieldName,
         /// Whether or not only distinct items should be counted
         distinct: bool,
     },
     SingleColumn {
         /// The column to apply the aggregation function to
-        column: String,
+        column: ndc_models::FieldName,
         /// Single column aggregate function name.
         function: T::AggregateFunction,
         result_type: Type<T::ScalarType>,
@@ -138,7 +138,7 @@ pub enum Aggregate<T: ConnectorTypes> {
 #[derive(Derivative)]
 #[derivative(Clone(bound = ""), Debug(bound = ""), PartialEq(bound = ""))]
 pub struct NestedObject<T: ConnectorTypes> {
-    pub fields: IndexMap<String, Field<T>>,
+    pub fields: IndexMap<ndc_models::FieldName, Field<T>>,
 }
 
 #[derive(Derivative)]
@@ -158,7 +158,7 @@ pub enum NestedField<T: ConnectorTypes> {
 #[derivative(Clone(bound = ""), Debug(bound = ""), PartialEq(bound = ""))]
 pub enum Field<T: ConnectorTypes> {
     Column {
-        column: String,
+        column: ndc_models::FieldName,
 
         /// When the type of the column is a (possibly-nullable) array or object,
         /// the caller can request a subset of the complete column data,
@@ -172,9 +172,9 @@ pub enum Field<T: ConnectorTypes> {
         /// The name of the relationship to follow for the subquery - this is the key in the
         /// [Query] relationships map in this module, it is **not** the key in the
         /// [ndc::QueryRequest] collection_relationships map.
-        relationship: String,
-        aggregates: Option<IndexMap<String, Aggregate<T>>>,
-        fields: Option<IndexMap<String, Field<T>>>,
+        relationship: ndc_models::RelationshipName,
+        aggregates: Option<IndexMap<ndc_models::FieldName, Aggregate<T>>>,
+        fields: Option<IndexMap<ndc_models::FieldName, Field<T>>>,
     },
 }
 
@@ -274,19 +274,19 @@ pub struct OrderByElement<T: ConnectorTypes> {
 pub enum OrderByTarget<T: ConnectorTypes> {
     Column {
         /// The name of the column
-        name: String,
+        name: ndc_models::FieldName,
 
         /// Path to a nested field within an object column
-        field_path: Option<Vec<String>>,
+        field_path: Option<Vec<ndc_models::FieldName>>,
 
         /// Any relationships to traverse to reach this column. These are translated from
         /// [ndc_models::OrderByElement] values in the [ndc_models::QueryRequest] to names of relation
         /// fields for the [QueryPlan].
-        path: Vec<String>,
+        path: Vec<ndc_models::RelationshipName>,
     },
     SingleColumnAggregate {
         /// The column to apply the aggregation function to
-        column: String,
+        column: ndc_models::FieldName,
         /// Single column aggregate function name.
         function: T::AggregateFunction,
 
@@ -295,13 +295,13 @@ pub enum OrderByTarget<T: ConnectorTypes> {
         /// Any relationships to traverse to reach this aggregate. These are translated from
         /// [ndc_models::OrderByElement] values in the [ndc_models::QueryRequest] to names of relation
         /// fields for the [QueryPlan].
-        path: Vec<String>,
+        path: Vec<ndc_models::RelationshipName>,
     },
     StarCountAggregate {
         /// Any relationships to traverse to reach this aggregate. These are translated from
         /// [ndc_models::OrderByElement] values in the [ndc_models::QueryRequest] to names of relation
         /// fields for the [QueryPlan].
-        path: Vec<String>,
+        path: Vec<ndc_models::RelationshipName>,
     },
 }
 
@@ -310,42 +310,42 @@ pub enum OrderByTarget<T: ConnectorTypes> {
 pub enum ComparisonTarget<T: ConnectorTypes> {
     Column {
         /// The name of the column
-        name: String,
+        name: ndc_models::FieldName,
 
         /// Path to a nested field within an object column
-        field_path: Option<Vec<String>>,
+        field_path: Option<Vec<ndc_models::FieldName>>,
 
         field_type: Type<T::ScalarType>,
 
         /// Any relationships to traverse to reach this column. These are translated from
         /// [ndc_models::PathElement] values in the [ndc_models::QueryRequest] to names of relation
         /// fields for the [QueryPlan].
-        path: Vec<String>,
+        path: Vec<ndc_models::RelationshipName>,
     },
     ColumnInScope {
         /// The name of the column
-        name: String,
+        name: ndc_models::FieldName,
 
         /// The named scope that identifies the collection to reference. This corresponds to the
         /// `scope` field of the [Query] type.
         scope: Scope,
 
         /// Path to a nested field within an object column
-        field_path: Option<Vec<String>>,
+        field_path: Option<Vec<ndc_models::FieldName>>,
 
         field_type: Type<T::ScalarType>,
     },
 }
 
 impl<T: ConnectorTypes> ComparisonTarget<T> {
-    pub fn column_name(&self) -> &str {
+    pub fn column_name(&self) -> &ndc_models::FieldName {
         match self {
             ComparisonTarget::Column { name, .. } => name,
             ComparisonTarget::ColumnInScope { name, .. } => name,
         }
     }
 
-    pub fn relationship_path(&self) -> &[String] {
+    pub fn relationship_path(&self) -> &[ndc_models::RelationshipName] {
         match self {
             ComparisonTarget::Column { path, .. } => path,
             ComparisonTarget::ColumnInScope { .. } => &[],
@@ -373,7 +373,7 @@ pub enum ComparisonValue<T: ConnectorTypes> {
         value_type: Type<T::ScalarType>,
     },
     Variable {
-        name: String,
+        name: ndc_models::VariableName,
         variable_type: Type<T::ScalarType>,
     },
 }
@@ -402,7 +402,7 @@ pub enum ExistsInCollection {
     Related {
         /// Key of the relation in the [Query] joins map. Relationships are scoped to the sub-query
         /// that defines the relation source.
-        relationship: String,
+        relationship: ndc_models::RelationshipName,
     },
     Unrelated {
         /// Key of the relation in the [QueryPlan] joins map. Unrelated collections are not scoped
diff --git a/crates/ndc-query-plan/src/type_system.rs b/crates/ndc-query-plan/src/type_system.rs
index b9adf6a9..36c0824a 100644
--- a/crates/ndc-query-plan/src/type_system.rs
+++ b/crates/ndc-query-plan/src/type_system.rs
@@ -1,3 +1,4 @@
+use ref_cast::RefCast;
 use std::collections::BTreeMap;
 
 use itertools::Itertools as _;
@@ -29,15 +30,13 @@ impl<S> Type<S> {
 pub struct ObjectType<ScalarType> {
     /// A type name may be tracked for error reporting. The name does not affect how query plans
     /// are generated.
-    pub name: Option<String>,
-    pub fields: BTreeMap<String, Type<ScalarType>>,
+    pub name: Option<ndc::ObjectTypeName>,
+    pub fields: BTreeMap<ndc::FieldName, Type<ScalarType>>,
 }
 
 impl<S> ObjectType<S> {
-    pub fn named_fields(&self) -> impl Iterator<Item = (&str, &Type<S>)> {
-        self.fields
-            .iter()
-            .map(|(name, field)| (name.as_ref(), field))
+    pub fn named_fields(&self) -> impl Iterator<Item = (&ndc::FieldName, &Type<S>)> {
+        self.fields.iter()
     }
 }
 
@@ -46,9 +45,9 @@ impl<S> ObjectType<S> {
 /// - query plan types are parameterized over the specific scalar type for a connector instead of
 ///   referencing scalar types by name
 pub fn inline_object_types<ScalarType>(
-    object_types: &BTreeMap<String, ndc::ObjectType>,
+    object_types: &BTreeMap<ndc::ObjectTypeName, ndc::ObjectType>,
     t: &ndc::Type,
-    lookup_scalar_type: fn(&str) -> Option<ScalarType>,
+    lookup_scalar_type: fn(&ndc::ScalarTypeName) -> Option<ScalarType>,
 ) -> Result<Type<ScalarType>, QueryPlanError> {
     let plan_type =
         match t {
@@ -67,28 +66,32 @@ pub fn inline_object_types<ScalarType>(
 }
 
 fn lookup_type<ScalarType>(
-    object_types: &BTreeMap<String, ndc::ObjectType>,
-    name: &str,
-    lookup_scalar_type: fn(&str) -> Option<ScalarType>,
+    object_types: &BTreeMap<ndc::ObjectTypeName, ndc::ObjectType>,
+    name: &ndc::TypeName,
+    lookup_scalar_type: fn(&ndc::ScalarTypeName) -> Option<ScalarType>,
 ) -> Result<plan::Type<ScalarType>, QueryPlanError> {
-    if let Some(scalar_type) = lookup_scalar_type(name) {
+    if let Some(scalar_type) = lookup_scalar_type(ndc::ScalarTypeName::ref_cast(name)) {
         return Ok(Type::Scalar(scalar_type));
     }
-    let object_type = lookup_object_type_helper(object_types, name, lookup_scalar_type)?;
+    let object_type = lookup_object_type_helper(
+        object_types,
+        ndc::ObjectTypeName::ref_cast(name),
+        lookup_scalar_type,
+    )?;
     Ok(Type::Object(object_type))
 }
 
 fn lookup_object_type_helper<ScalarType>(
-    object_types: &BTreeMap<String, ndc::ObjectType>,
-    name: &str,
-    lookup_scalar_type: fn(&str) -> Option<ScalarType>,
+    object_types: &BTreeMap<ndc::ObjectTypeName, ndc::ObjectType>,
+    name: &ndc::ObjectTypeName,
+    lookup_scalar_type: fn(&ndc::ScalarTypeName) -> Option<ScalarType>,
 ) -> Result<plan::ObjectType<ScalarType>, QueryPlanError> {
     let object_type = object_types
         .get(name)
         .ok_or_else(|| QueryPlanError::UnknownObjectType(name.to_string()))?;
 
     let plan_object_type = plan::ObjectType {
-        name: Some(name.to_owned()),
+        name: Some(name.clone()),
         fields: object_type
             .fields
             .iter()
@@ -104,9 +107,9 @@ fn lookup_object_type_helper<ScalarType>(
 }
 
 pub fn lookup_object_type<ScalarType>(
-    object_types: &BTreeMap<String, ndc::ObjectType>,
-    name: &str,
-    lookup_scalar_type: fn(&str) -> Option<ScalarType>,
+    object_types: &BTreeMap<ndc::ObjectTypeName, ndc::ObjectType>,
+    name: &ndc::ObjectTypeName,
+    lookup_scalar_type: fn(&ndc::ScalarTypeName) -> Option<ScalarType>,
 ) -> Result<plan::ObjectType<ScalarType>, QueryPlanError> {
     lookup_object_type_helper(object_types, name, lookup_scalar_type)
 }
diff --git a/crates/ndc-test-helpers/src/aggregates.rs b/crates/ndc-test-helpers/src/aggregates.rs
index 6579273d..212222c1 100644
--- a/crates/ndc-test-helpers/src/aggregates.rs
+++ b/crates/ndc-test-helpers/src/aggregates.rs
@@ -4,8 +4,8 @@ macro_rules! column_aggregate {
         (
             $name,
             $crate::ndc_models::Aggregate::SingleColumn {
-                column: $column.to_owned(),
-                function: $function.to_owned(),
+                column: $column.into(),
+                function: $function.into(),
                 field_path: None,
             },
         )
@@ -25,7 +25,7 @@ macro_rules! column_count_aggregate {
         (
             $name,
             $crate::ndc_models::Aggregate::ColumnCount {
-                column: $column.to_owned(),
+                column: $column.into(),
                 distinct: $distinct.to_owned(),
                 field_path: None,
             },
diff --git a/crates/ndc-test-helpers/src/collection_info.rs b/crates/ndc-test-helpers/src/collection_info.rs
index 4b41d802..3e042711 100644
--- a/crates/ndc-test-helpers/src/collection_info.rs
+++ b/crates/ndc-test-helpers/src/collection_info.rs
@@ -2,16 +2,16 @@ use std::{collections::BTreeMap, fmt::Display};
 
 use ndc_models::{CollectionInfo, ObjectField, ObjectType, Type, UniquenessConstraint};
 
-pub fn collection(name: impl Display + Clone) -> (String, CollectionInfo) {
+pub fn collection(name: impl Display + Clone) -> (ndc_models::CollectionName, CollectionInfo) {
     let coll = CollectionInfo {
-        name: name.to_string(),
+        name: name.to_string().into(),
         description: None,
         arguments: Default::default(),
-        collection_type: name.to_string(),
+        collection_type: name.to_string().into(),
         uniqueness_constraints: make_primary_key_uniqueness_constraint(name.clone()),
         foreign_keys: Default::default(),
     };
-    (name.to_string(), coll)
+    (name.to_string().into(), coll)
 }
 
 pub fn make_primary_key_uniqueness_constraint(
@@ -20,7 +20,7 @@ pub fn make_primary_key_uniqueness_constraint(
     [(
         format!("{collection_name}_id"),
         UniquenessConstraint {
-            unique_columns: vec!["_id".to_owned()],
+            unique_columns: vec!["_id".to_owned().into()],
         },
     )]
     .into()
diff --git a/crates/ndc-test-helpers/src/comparison_target.rs b/crates/ndc-test-helpers/src/comparison_target.rs
index b8f9533f..41463113 100644
--- a/crates/ndc-test-helpers/src/comparison_target.rs
+++ b/crates/ndc-test-helpers/src/comparison_target.rs
@@ -2,28 +2,28 @@
 macro_rules! target {
     ($column:literal) => {
         $crate::ndc_models::ComparisonTarget::Column {
-            name: $column.to_owned(),
+            name: $column.into(),
             field_path: None,
             path: vec![],
         }
     };
     ($column:literal, field_path:$field_path:expr $(,)?) => {
         $crate::ndc_models::ComparisonTarget::Column {
-            name: $column.to_owned(),
+            name: $column.into(),
             field_path: $field_path.into_iter().map(|x| x.into()).collect(),
             path: vec![],
         }
     };
     ($column:literal, relations:$path:expr $(,)?) => {
         $crate::ndc_models::ComparisonTarget::Column {
-            name: $column.to_owned(),
+            name: $column.into(),
             field_path: None,
             path: $path.into_iter().map(|x| x.into()).collect(),
         }
     };
     ($column:literal, field_path:$field_path:expr, relations:$path:expr $(,)?) => {
         $crate::ndc_models::ComparisonTarget::Column {
-            name: $column.to_owned(),
+            name: $column.into(),
             // field_path: $field_path.into_iter().map(|x| x.into()).collect(),
             path: $path.into_iter().map(|x| x.into()).collect(),
         }
@@ -38,7 +38,7 @@ where
     S: ToString,
 {
     ndc_models::ComparisonTarget::RootCollectionColumn {
-        name: name.to_string(),
+        name: name.to_string().into(),
         field_path: None,
     }
 }
diff --git a/crates/ndc-test-helpers/src/comparison_value.rs b/crates/ndc-test-helpers/src/comparison_value.rs
index 0d233bb5..350378e1 100644
--- a/crates/ndc-test-helpers/src/comparison_value.rs
+++ b/crates/ndc-test-helpers/src/comparison_value.rs
@@ -20,7 +20,7 @@ macro_rules! value {
 macro_rules! variable {
     ($variable:ident) => {
         $crate::ndc_models::ComparisonValue::Variable {
-            name: stringify!($variable).to_owned(),
+            name: stringify!($variable).into(),
         }
     };
     ($variable:expr) => {
diff --git a/crates/ndc-test-helpers/src/exists_in_collection.rs b/crates/ndc-test-helpers/src/exists_in_collection.rs
index 5208086e..e13826c6 100644
--- a/crates/ndc-test-helpers/src/exists_in_collection.rs
+++ b/crates/ndc-test-helpers/src/exists_in_collection.rs
@@ -2,13 +2,13 @@
 macro_rules! related {
     ($rel:literal) => {
         $crate::ndc_models::ExistsInCollection::Related {
-            relationship: $rel.to_owned(),
+            relationship: $rel.into(),
             arguments: Default::default(),
         }
     };
     ($rel:literal, $args:expr $(,)?) => {
         $crate::ndc_models::ExistsInCollection::Related {
-            relationship: $rel.to_owned(),
+            relationship: $rel.into(),
             arguments: $args.into_iter().map(|x| x.into()).collect(),
         }
     };
@@ -18,13 +18,13 @@ macro_rules! related {
 macro_rules! unrelated {
     ($coll:literal) => {
         $crate::ndc_models::ExistsInCollection::Unrelated {
-            collection: $coll.to_owned(),
+            collection: $coll.into(),
             arguments: Default::default(),
         }
     };
     ($coll:literal, $args:expr $(,)?) => {
         $crate::ndc_models::ExistsInCollection::Related {
-            collection: $coll.to_owned(),
+            collection: $coll.into(),
             arguments: $args.into_iter().map(|x| x.into()).collect(),
         }
     };
diff --git a/crates/ndc-test-helpers/src/expressions.rs b/crates/ndc-test-helpers/src/expressions.rs
index 26c69e5f..6b35ae2a 100644
--- a/crates/ndc-test-helpers/src/expressions.rs
+++ b/crates/ndc-test-helpers/src/expressions.rs
@@ -39,7 +39,7 @@ where
 {
     Expression::BinaryComparisonOperator {
         column: op1,
-        operator: oper.to_string(),
+        operator: oper.to_string().into(),
         value: op2,
     }
 }
@@ -50,7 +50,7 @@ where
 {
     Expression::BinaryComparisonOperator {
         column: op1,
-        operator: "_in".to_owned(),
+        operator: "_in".into(),
         value: ComparisonValue::Scalar {
             value: values.into_iter().collect(),
         },
diff --git a/crates/ndc-test-helpers/src/field.rs b/crates/ndc-test-helpers/src/field.rs
index 18cee830..b1cae0a6 100644
--- a/crates/ndc-test-helpers/src/field.rs
+++ b/crates/ndc-test-helpers/src/field.rs
@@ -4,7 +4,7 @@ macro_rules! field {
         (
             $name,
             $crate::ndc_models::Field::Column {
-                column: $name.to_owned(),
+                column: $name.into(),
                 arguments: Default::default(),
                 fields: None,
             },
@@ -14,7 +14,7 @@ macro_rules! field {
         (
             $name,
             $crate::ndc_models::Field::Column {
-                column: $column_name.to_owned(),
+                column: $column_name.into(),
                 arguments: Default::default(),
                 fields: None,
             },
@@ -24,7 +24,7 @@ macro_rules! field {
         (
             $name,
             $crate::ndc_models::Field::Column {
-                column: $column_name.to_owned(),
+                column: $column_name.into(),
                 arguments: Default::default(),
                 fields: Some($fields.into()),
             },
@@ -38,7 +38,7 @@ macro_rules! object {
         $crate::ndc_models::NestedField::Object($crate::ndc_models::NestedObject {
             fields: $fields
                 .into_iter()
-                .map(|(name, field)| (name.to_owned(), field))
+                .map(|(name, field)| (name.into(), field))
                 .collect(),
         })
     };
@@ -60,7 +60,7 @@ macro_rules! relation_field {
             $name,
             $crate::ndc_models::Field::Relationship {
                 query: Box::new($crate::query().into()),
-                relationship: $relationship.to_owned(),
+                relationship: $relationship.into(),
                 arguments: Default::default(),
             },
         )
@@ -70,7 +70,7 @@ macro_rules! relation_field {
             $name,
             $crate::ndc_models::Field::Relationship {
                 query: Box::new($query.into()),
-                relationship: $relationship.to_owned(),
+                relationship: $relationship.into(),
                 arguments: Default::default(),
             },
         )
diff --git a/crates/ndc-test-helpers/src/lib.rs b/crates/ndc-test-helpers/src/lib.rs
index 1859cf6c..1e30c2ca 100644
--- a/crates/ndc-test-helpers/src/lib.rs
+++ b/crates/ndc-test-helpers/src/lib.rs
@@ -39,11 +39,11 @@ pub use type_helpers::*;
 
 #[derive(Clone, Debug, Default)]
 pub struct QueryRequestBuilder {
-    collection: Option<String>,
+    collection: Option<ndc_models::CollectionName>,
     query: Option<Query>,
-    arguments: Option<BTreeMap<String, Argument>>,
-    collection_relationships: Option<BTreeMap<String, Relationship>>,
-    variables: Option<Vec<BTreeMap<String, serde_json::Value>>>,
+    arguments: Option<BTreeMap<ndc_models::ArgumentName, Argument>>,
+    collection_relationships: Option<BTreeMap<ndc_models::RelationshipName, Relationship>>,
+    variables: Option<Vec<BTreeMap<ndc_models::VariableName, serde_json::Value>>>,
 }
 
 pub fn query_request() -> QueryRequestBuilder {
@@ -62,7 +62,7 @@ impl QueryRequestBuilder {
     }
 
     pub fn collection(mut self, collection: &str) -> Self {
-        self.collection = Some(collection.to_owned());
+        self.collection = Some(collection.to_owned().into());
         self
     }
 
@@ -75,7 +75,7 @@ impl QueryRequestBuilder {
         self.arguments = Some(
             arguments
                 .into_iter()
-                .map(|(name, arg)| (name.to_owned(), arg))
+                .map(|(name, arg)| (name.to_owned().into(), arg))
                 .collect(),
         );
         self
@@ -88,7 +88,7 @@ impl QueryRequestBuilder {
         self.collection_relationships = Some(
             relationships
                 .into_iter()
-                .map(|(name, r)| (name.to_string(), r.into()))
+                .map(|(name, r)| (name.to_string().into(), r.into()))
                 .collect(),
         );
         self
@@ -106,7 +106,7 @@ impl QueryRequestBuilder {
                 .map(|var_map| {
                     var_map
                         .into_iter()
-                        .map(|(name, value)| (name.to_string(), value.into()))
+                        .map(|(name, value)| (name.to_string().into(), value.into()))
                         .collect()
                 })
                 .collect(),
@@ -133,8 +133,8 @@ impl From<QueryRequestBuilder> for QueryRequest {
 
 #[derive(Clone, Debug, Default)]
 pub struct QueryBuilder {
-    aggregates: Option<IndexMap<String, Aggregate>>,
-    fields: Option<IndexMap<String, Field>>,
+    aggregates: Option<IndexMap<ndc_models::FieldName, Aggregate>>,
+    fields: Option<IndexMap<ndc_models::FieldName, Field>>,
     limit: Option<u32>,
     offset: Option<u32>,
     order_by: Option<OrderBy>,
@@ -161,7 +161,7 @@ impl QueryBuilder {
         self.fields = Some(
             fields
                 .into_iter()
-                .map(|(name, field)| (name.to_owned(), field))
+                .map(|(name, field)| (name.to_owned().into(), field))
                 .collect(),
         );
         self
@@ -171,7 +171,7 @@ impl QueryBuilder {
         self.aggregates = Some(
             aggregates
                 .into_iter()
-                .map(|(name, aggregate)| (name.to_owned(), aggregate))
+                .map(|(name, aggregate)| (name.to_owned().into(), aggregate))
                 .collect(),
         );
         self
diff --git a/crates/ndc-test-helpers/src/object_type.rs b/crates/ndc-test-helpers/src/object_type.rs
index 58758525..01feb919 100644
--- a/crates/ndc-test-helpers/src/object_type.rs
+++ b/crates/ndc-test-helpers/src/object_type.rs
@@ -11,7 +11,7 @@ pub fn object_type(
             .into_iter()
             .map(|(name, field_type)| {
                 (
-                    name.to_string(),
+                    name.to_string().into(),
                     ObjectField {
                         description: Default::default(),
                         arguments: BTreeMap::new(),
diff --git a/crates/ndc-test-helpers/src/path_element.rs b/crates/ndc-test-helpers/src/path_element.rs
index d0ee34e6..b0c89d5b 100644
--- a/crates/ndc-test-helpers/src/path_element.rs
+++ b/crates/ndc-test-helpers/src/path_element.rs
@@ -4,19 +4,19 @@ use ndc_models::{Expression, PathElement, RelationshipArgument};
 
 #[derive(Clone, Debug)]
 pub struct PathElementBuilder {
-    relationship: String,
-    arguments: Option<BTreeMap<String, RelationshipArgument>>,
+    relationship: ndc_models::RelationshipName,
+    arguments: Option<BTreeMap<ndc_models::ArgumentName, RelationshipArgument>>,
     predicate: Option<Box<Expression>>,
 }
 
-pub fn path_element(relationship: &str) -> PathElementBuilder {
+pub fn path_element(relationship: ndc_models::RelationshipName) -> PathElementBuilder {
     PathElementBuilder::new(relationship)
 }
 
 impl PathElementBuilder {
-    pub fn new(relationship: &str) -> Self {
+    pub fn new(relationship: ndc_models::RelationshipName) -> Self {
         PathElementBuilder {
-            relationship: relationship.to_owned(),
+            relationship,
             arguments: None,
             predicate: None,
         }
diff --git a/crates/ndc-test-helpers/src/query_response.rs b/crates/ndc-test-helpers/src/query_response.rs
index 41c39545..72970bb2 100644
--- a/crates/ndc-test-helpers/src/query_response.rs
+++ b/crates/ndc-test-helpers/src/query_response.rs
@@ -43,8 +43,8 @@ impl From<QueryResponseBuilder> for QueryResponse {
 
 #[derive(Clone, Debug, Default)]
 pub struct RowSetBuilder {
-    aggregates: IndexMap<String, serde_json::Value>,
-    rows: Vec<IndexMap<String, RowFieldValue>>,
+    aggregates: IndexMap<ndc_models::FieldName, serde_json::Value>,
+    rows: Vec<IndexMap<ndc_models::FieldName, RowFieldValue>>,
 }
 
 impl RowSetBuilder {
@@ -59,7 +59,7 @@ impl RowSetBuilder {
         self.aggregates.extend(
             aggregates
                 .into_iter()
-                .map(|(k, v)| (k.to_string(), v.into())),
+                .map(|(k, v)| (k.to_string().into(), v.into())),
         );
         self
     }
@@ -72,7 +72,7 @@ impl RowSetBuilder {
     ) -> Self {
         self.rows.extend(rows.into_iter().map(|r| {
             r.into_iter()
-                .map(|(k, v)| (k.to_string(), RowFieldValue(v.into())))
+                .map(|(k, v)| (k.to_string().into(), RowFieldValue(v.into())))
                 .collect()
         }));
         self
@@ -84,7 +84,7 @@ impl RowSetBuilder {
     ) -> Self {
         self.rows.push(
             row.into_iter()
-                .map(|(k, v)| (k.to_string(), RowFieldValue(v.into())))
+                .map(|(k, v)| (k.to_string().into(), RowFieldValue(v.into())))
                 .collect(),
         );
         self
diff --git a/crates/ndc-test-helpers/src/relationships.rs b/crates/ndc-test-helpers/src/relationships.rs
index bdf9853c..6166e809 100644
--- a/crates/ndc-test-helpers/src/relationships.rs
+++ b/crates/ndc-test-helpers/src/relationships.rs
@@ -4,10 +4,10 @@ use ndc_models::{Relationship, RelationshipArgument, RelationshipType};
 
 #[derive(Clone, Debug)]
 pub struct RelationshipBuilder {
-    column_mapping: BTreeMap<String, String>,
+    column_mapping: BTreeMap<ndc_models::FieldName, ndc_models::FieldName>,
     relationship_type: RelationshipType,
-    target_collection: String,
-    arguments: BTreeMap<String, RelationshipArgument>,
+    target_collection: ndc_models::CollectionName,
+    arguments: BTreeMap<ndc_models::ArgumentName, RelationshipArgument>,
 }
 
 pub fn relationship<const S: usize>(
@@ -22,10 +22,10 @@ impl RelationshipBuilder {
         RelationshipBuilder {
             column_mapping: column_mapping
                 .into_iter()
-                .map(|(source, target)| (source.to_owned(), target.to_owned()))
+                .map(|(source, target)| (source.to_owned().into(), target.to_owned().into()))
                 .collect(),
             relationship_type: RelationshipType::Array,
-            target_collection: target.to_owned(),
+            target_collection: target.to_owned().into(),
             arguments: Default::default(),
         }
     }
@@ -40,7 +40,10 @@ impl RelationshipBuilder {
         self
     }
 
-    pub fn arguments(mut self, arguments: BTreeMap<String, RelationshipArgument>) -> Self {
+    pub fn arguments(
+        mut self,
+        arguments: BTreeMap<ndc_models::ArgumentName, RelationshipArgument>,
+    ) -> Self {
         self.arguments = arguments;
         self
     }
diff --git a/crates/ndc-test-helpers/src/type_helpers.rs b/crates/ndc-test-helpers/src/type_helpers.rs
index 025ab880..207f4652 100644
--- a/crates/ndc-test-helpers/src/type_helpers.rs
+++ b/crates/ndc-test-helpers/src/type_helpers.rs
@@ -8,7 +8,7 @@ pub fn array_of(t: impl Into<Type>) -> Type {
 
 pub fn named_type(name: impl ToString) -> Type {
     Type::Named {
-        name: name.to_string(),
+        name: name.to_string().into(),
     }
 }
 
diff --git a/crates/test-helpers/src/arb_plan_type.rs b/crates/test-helpers/src/arb_plan_type.rs
index b878557a..0ffe5ac1 100644
--- a/crates/test-helpers/src/arb_plan_type.rs
+++ b/crates/test-helpers/src/arb_plan_type.rs
@@ -12,9 +12,12 @@ pub fn arb_plan_type() -> impl Strategy<Value = Type<MongoScalarType>> {
             inner.clone().prop_map(|t| Type::Nullable(Box::new(t))),
             (
                 any::<Option<String>>(),
-                btree_map(any::<String>(), inner, 1..=10)
+                btree_map(any::<String>().prop_map_into(), inner, 1..=10)
             )
-                .prop_map(|(name, fields)| Type::Object(ObjectType { name, fields }))
+                .prop_map(|(name, fields)| Type::Object(ObjectType {
+                    name: name.map(|n| n.into()),
+                    fields
+                }))
         ]
     })
 }
diff --git a/flake.lock b/flake.lock
index 66a1ea0b..6192f37f 100644
--- a/flake.lock
+++ b/flake.lock
@@ -3,11 +3,11 @@
     "advisory-db": {
       "flake": false,
       "locked": {
-        "lastModified": 1712168594,
-        "narHash": "sha256-1Yh+vafNq19JDfmpknkWq11AkcQLPmFZ8X6YJZT5r7o=",
+        "lastModified": 1720572893,
+        "narHash": "sha256-EQfU1yMnebn7LoJNjjsQimyuWwz+2YzazqUZu8aX/r4=",
         "owner": "rustsec",
         "repo": "advisory-db",
-        "rev": "0bc9a77248be5cb5f2b51fe6aba8ba451d74c6bb",
+        "rev": "97a2dc75838f19a5fd63dc3f8e3f57e0c4c8cfe6",
         "type": "github"
       },
       "original": {
@@ -26,11 +26,11 @@
         ]
       },
       "locked": {
-        "lastModified": 1709606645,
-        "narHash": "sha256-yObjAl8deNvx1uIfQn7/vkB9Rnr0kqTo1HVrsk46l30=",
+        "lastModified": 1720147808,
+        "narHash": "sha256-hlWEQGUbIwYb+vnd8egzlW/P++yKu3HjV/rOdOPVank=",
         "owner": "hercules-ci",
         "repo": "arion",
-        "rev": "d2d48c9ec304ac80c84ede138b8c6f298d07d995",
+        "rev": "236f9dd82d6ef6a2d9987c7a7df3e75f1bc8b318",
         "type": "github"
       },
       "original": {
@@ -46,11 +46,11 @@
         ]
       },
       "locked": {
-        "lastModified": 1712180168,
-        "narHash": "sha256-sYe00cK+kKnQlVo1wUIZ5rZl9x8/r3djShUqNgfjnM4=",
+        "lastModified": 1720546058,
+        "narHash": "sha256-iU2yVaPIZm5vMGdlT0+57vdB/aPq/V5oZFBRwYw+HBM=",
         "owner": "ipetkov",
         "repo": "crane",
-        "rev": "06a9ff255c1681299a87191c2725d9d579f28b82",
+        "rev": "2d83156f23c43598cf44e152c33a59d3892f8b29",
         "type": "github"
       },
       "original": {
@@ -82,11 +82,11 @@
         ]
       },
       "locked": {
-        "lastModified": 1709336216,
-        "narHash": "sha256-Dt/wOWeW6Sqm11Yh+2+t0dfEWxoMxGBvv3JpIocFl9E=",
+        "lastModified": 1719994518,
+        "narHash": "sha256-pQMhCCHyQGRzdfAkdJ4cIWiw+JNuWsTX7f0ZYSyz0VY=",
         "owner": "hercules-ci",
         "repo": "flake-parts",
-        "rev": "f7b3c975cf067e56e7cda6cb098ebe3fb4d74ca2",
+        "rev": "9227223f6d922fee3c7b190b2cc238a99527bbb7",
         "type": "github"
       },
       "original": {
@@ -104,11 +104,11 @@
         ]
       },
       "locked": {
-        "lastModified": 1701473968,
-        "narHash": "sha256-YcVE5emp1qQ8ieHUnxt1wCZCC3ZfAS+SRRWZ2TMda7E=",
+        "lastModified": 1712014858,
+        "narHash": "sha256-sB4SWl2lX95bExY2gMFG5HIzvva5AVMJd4Igm+GpZNw=",
         "owner": "hercules-ci",
         "repo": "flake-parts",
-        "rev": "34fed993f1674c8d06d58b37ce1e0fe5eebcb9f5",
+        "rev": "9126214d0a59633752a136528f5f3b9aa8565b7d",
         "type": "github"
       },
       "original": {
@@ -116,24 +116,6 @@
         "type": "indirect"
       }
     },
-    "flake-utils": {
-      "inputs": {
-        "systems": "systems"
-      },
-      "locked": {
-        "lastModified": 1705309234,
-        "narHash": "sha256-uNRRNRKmJyCRC/8y1RqBkqWBLM034y4qN7EprSdmgyA=",
-        "owner": "numtide",
-        "repo": "flake-utils",
-        "rev": "1ef2e671c3b0c19053962c07dbda38332dcebf26",
-        "type": "github"
-      },
-      "original": {
-        "owner": "numtide",
-        "repo": "flake-utils",
-        "type": "github"
-      }
-    },
     "graphql-engine-source": {
       "flake": false,
       "locked": {
@@ -175,11 +157,11 @@
         ]
       },
       "locked": {
-        "lastModified": 1708547820,
-        "narHash": "sha256-xU/KC1PWqq5zL9dQ9wYhcdgxAwdeF/dJCLPH3PNZEBg=",
+        "lastModified": 1719226092,
+        "narHash": "sha256-YNkUMcCUCpnULp40g+svYsaH1RbSEj6s4WdZY/SHe38=",
         "owner": "hercules-ci",
         "repo": "hercules-ci-effects",
-        "rev": "0ca27bd58e4d5be3135a4bef66b582e57abe8f4a",
+        "rev": "11e4b8dc112e2f485d7c97e1cee77f9958f498f5",
         "type": "github"
       },
       "original": {
@@ -190,11 +172,11 @@
     },
     "nixpkgs": {
       "locked": {
-        "lastModified": 1712163089,
-        "narHash": "sha256-Um+8kTIrC19vD4/lUCN9/cU9kcOsD1O1m+axJqQPyMM=",
+        "lastModified": 1720542800,
+        "narHash": "sha256-ZgnNHuKV6h2+fQ5LuqnUaqZey1Lqqt5dTUAiAnqH0QQ=",
         "owner": "NixOS",
         "repo": "nixpkgs",
-        "rev": "fd281bd6b7d3e32ddfa399853946f782553163b5",
+        "rev": "feb2849fdeb70028c70d73b848214b00d324a497",
         "type": "github"
       },
       "original": {
@@ -213,22 +195,21 @@
         "graphql-engine-source": "graphql-engine-source",
         "nixpkgs": "nixpkgs",
         "rust-overlay": "rust-overlay",
-        "systems": "systems_2"
+        "systems": "systems"
       }
     },
     "rust-overlay": {
       "inputs": {
-        "flake-utils": "flake-utils",
         "nixpkgs": [
           "nixpkgs"
         ]
       },
       "locked": {
-        "lastModified": 1712196778,
-        "narHash": "sha256-SOiwCr2HtmYpw8OvQQVRPtiCBWwndbIoPqtsamZK3J8=",
+        "lastModified": 1720577957,
+        "narHash": "sha256-RZuzLdB/8FaXaSzEoWLg3au/mtbuH7MGn2LmXUKT62g=",
         "owner": "oxalica",
         "repo": "rust-overlay",
-        "rev": "20e7895d1873cc64c14a9f024a8e04f5824bed28",
+        "rev": "a434177dfcc53bf8f1f348a3c39bfb336d760286",
         "type": "github"
       },
       "original": {
@@ -251,21 +232,6 @@
         "repo": "default",
         "type": "github"
       }
-    },
-    "systems_2": {
-      "locked": {
-        "lastModified": 1681028828,
-        "narHash": "sha256-Vy1rq5AaRuLzOxct8nz4T6wlgyUR7zLU309k9mBC768=",
-        "owner": "nix-systems",
-        "repo": "default",
-        "rev": "da67096a3b9bf56a91d16901293e51ba5b49a27e",
-        "type": "github"
-      },
-      "original": {
-        "owner": "nix-systems",
-        "repo": "default",
-        "type": "github"
-      }
     }
   },
   "root": "root",