Handle :attribute/_reverse in transactor. Fixes #187. r=rnewman

db/src/db.rs

@@ -2085,6 +2085,164 @@ mod tests {
 
         // Verify that we can explode map notation with nested maps, even if the inner map would be
         // dangling, if we give a :db/id explicitly.
-        assert_transact!(conn, "[{:test/dangling {:db/id \"t\" :test/many 11}}]");
+        assert_transact!(conn, "[{:test/dangling {:db/id \"t\" :test/many 12}}]");
+    }
+
+    #[test]
+    fn test_explode_reversed_notation() {
+        let mut conn = TestConn::default();
+
+        // Start by installing a few attributes.
+        assert_transact!(conn, "[[:db/add 111 :db/ident :test/many]
+                                 [:db/add 111 :db/valueType :db.type/long]
+                                 [:db/add 111 :db/cardinality :db.cardinality/many]
+                                 [:db/add 222 :db/ident :test/component]
+                                 [:db/add 222 :db/isComponent true]
+                                 [:db/add 222 :db/valueType :db.type/ref]
+                                 [:db/add 333 :db/ident :test/unique]
+                                 [:db/add 333 :db/unique :db.unique/identity]
+                                 [:db/add 333 :db/index true]
+                                 [:db/add 333 :db/valueType :db.type/long]
+                                 [:db/add 444 :db/ident :test/dangling]
+                                 [:db/add 444 :db/valueType :db.type/ref]]");
+
+        // Check that we can explode direct reversed notation, entids.
+        let report = assert_transact!(conn, "[[:db/add 100 :test/_dangling 200]]");
+        assert_matches!(conn.last_transaction(),
+                        "[[200 :test/dangling 100 ?tx true]
+                          [?tx :db/txInstant ?ms ?tx true]]");
+        assert_matches!(tempids(&report),
+                        "{}");
+
+        // Check that we can explode direct reversed notation, idents.
+        let report = assert_transact!(conn, "[[:db/add :test/many :test/_dangling :test/unique]]");
+        assert_matches!(conn.last_transaction(),
+                        "[[333 :test/dangling :test/many ?tx true]
+                          [?tx :db/txInstant ?ms ?tx true]]");
+        assert_matches!(tempids(&report),
+                        "{}");
+
+        // Check that we can explode direct reversed notation, tempids.
+        let report = assert_transact!(conn, "[[:db/add \"s\" :test/_dangling \"t\"]]");
+        assert_matches!(conn.last_transaction(),
+                        "[[65537 :test/dangling 65536 ?tx true]
+                          [?tx :db/txInstant ?ms ?tx true]]");
+        // This is implementation specific, but it should be deterministic.
+        assert_matches!(tempids(&report),
+                        "{\"s\" 65536
+                          \"t\" 65537}");
+
+        // Check that we can explode reversed notation in map notation without :db/id.
+        let report = assert_transact!(conn, "[{:test/_dangling 501}
+                                              {:test/_dangling :test/many}
+                                              {:test/_dangling \"t\"}]");
+        assert_matches!(conn.last_transaction(),
+                        "[[111 :test/dangling ?e1 ?tx true]
+                          [501 :test/dangling ?e2 ?tx true]
+                          [65538 :test/dangling ?e3 ?tx true]
+                          [?tx :db/txInstant ?ms ?tx true]]");
+        assert_matches!(tempids(&report),
+                        "{\"t\" 65538}");
+
+        // Check that we can explode reversed notation in map notation with :db/id, entid.
+        let report = assert_transact!(conn, "[{:db/id 600 :test/_dangling 601}]");
+        assert_matches!(conn.last_transaction(),
+                        "[[601 :test/dangling 600 ?tx true]
+                          [?tx :db/txInstant ?ms ?tx true]]");
+        assert_matches!(tempids(&report),
+                        "{}");
+
+        // Check that we can explode reversed notation in map notation with :db/id, ident.
+        let report = assert_transact!(conn, "[{:db/id :test/component :test/_dangling :test/component}]");
+        assert_matches!(conn.last_transaction(),
+                        "[[222 :test/dangling :test/component ?tx true]
+                          [?tx :db/txInstant ?ms ?tx true]]");
+        assert_matches!(tempids(&report),
+                        "{}");
+
+        // Check that we can explode reversed notation in map notation with :db/id, tempid.
+        let report = assert_transact!(conn, "[{:db/id \"s\" :test/_dangling \"t\"}]");
+        assert_matches!(conn.last_transaction(),
+                        "[[65543 :test/dangling 65542 ?tx true]
+                          [?tx :db/txInstant ?ms ?tx true]]");
+        // This is implementation specific, but it should be deterministic.
+        assert_matches!(tempids(&report),
+                        "{\"s\" 65542
+                          \"t\" 65543}");
+
+        // Check that we can use the same attribute in both forward and backward form in the same
+        // transaction.
+        let report = assert_transact!(conn, "[[:db/add 888 :test/dangling 889]
+                                              [:db/add 888 :test/_dangling 889]]");
+        assert_matches!(conn.last_transaction(),
+                        "[[888 :test/dangling 889 ?tx true]
+                          [889 :test/dangling 888 ?tx true]
+                          [?tx :db/txInstant ?ms ?tx true]]");
+        assert_matches!(tempids(&report),
+                        "{}");
+
+        // Check that we can use the same attribute in both forward and backward form in the same
+        // transaction in map notation.
+        let report = assert_transact!(conn, "[{:db/id 998 :test/dangling 999 :test/_dangling 999}]");
+        assert_matches!(conn.last_transaction(),
+                        "[[998 :test/dangling 999 ?tx true]
+                          [999 :test/dangling 998 ?tx true]
+                          [?tx :db/txInstant ?ms ?tx true]]");
+        assert_matches!(tempids(&report),
+                        "{}");
+
+    }
+
+    #[test]
+    fn test_explode_reversed_notation_errors() {
+        let mut conn = TestConn::default();
+
+        // Start by installing a few attributes.
+        assert_transact!(conn, "[[:db/add 111 :db/ident :test/many]
+                                 [:db/add 111 :db/valueType :db.type/long]
+                                 [:db/add 111 :db/cardinality :db.cardinality/many]
+                                 [:db/add 222 :db/ident :test/component]
+                                 [:db/add 222 :db/isComponent true]
+                                 [:db/add 222 :db/valueType :db.type/ref]
+                                 [:db/add 333 :db/ident :test/unique]
+                                 [:db/add 333 :db/unique :db.unique/identity]
+                                 [:db/add 333 :db/index true]
+                                 [:db/add 333 :db/valueType :db.type/long]
+                                 [:db/add 444 :db/ident :test/dangling]
+                                 [:db/add 444 :db/valueType :db.type/ref]]");
+
+        // `tx-parser` should fail to parse direct reverse notation with nested value maps and
+        // nested value vectors, so we only test things that "get through" to the map notation
+        // dynamic processor here.
+
+        // Verify that we can't explode reverse notation in map notation with nested value maps.
+        assert_transact!(conn,
+                         "[{:test/_dangling {:test/many 14}}]",
+                         Err("not yet implemented: Cannot explode map notation value in :attr/_reversed notation for attribute 444"));
+
+        // Verify that we can't explode reverse notation in map notation with nested value vectors.
+        assert_transact!(conn,
+                         "[{:test/_dangling [:test/many]}]",
+                         Err("not yet implemented: Cannot explode vector value in :attr/_reversed notation for attribute 444"));
+
+        // Verify that we can't use reverse notation with non-:db.type/ref attributes.
+        assert_transact!(conn,
+                         "[{:test/_unique 500}]",
+                         Err("not yet implemented: Cannot use :attr/_reversed notation for attribute 333 that is not :db/valueType :db.type/ref"));
+
+        // Verify that we can't use reverse notation with unrecognized attributes.
+        assert_transact!(conn,
+                         "[{:test/_unknown 500}]",
+                         Err("no entid found for ident: :test/unknown")); // TODO: make this error reference the original :test/_unknown.
+
+        // Verify that we can't use reverse notation with bad value types: here, an unknown keyword
+        // that can't be coerced to a ref.
+        assert_transact!(conn,
+                         "[{:test/_dangling :test/unknown}]",
+                         Err("no entid found for ident: :test/unknown"));
+        // And here, a float.
+        assert_transact!(conn,
+                         "[{:test/_dangling 1.23}]",
+                         Err("EDN value \'1.23\' is not the expected Mentat value type Ref"));
     }
 }

db/src/internal_types.rs

@@ -39,6 +39,27 @@ pub enum Either<L, R> {
     Right(R),
 }
 
+// Cribbed from https://github.com/bluss/either/blob/f793721f3fdeb694f009e731b23a2858286bc0d6/src/lib.rs#L219-L259.
+impl<L, R> Either<L, R> {
+    pub fn map_left<F, M>(self, f: F) -> Either<M, R>
+        where F: FnOnce(L) -> M
+    {
+        match self {
+            Left(l) => Left(f(l)),
+            Right(r) => Right(r),
+        }
+    }
+
+    pub fn map_right<F, S>(self, f: F) -> Either<L, S>
+        where F: FnOnce(R) -> S
+    {
+        match self {
+            Left(l) => Left(l),
+            Right(r) => Right(f(r)),
+        }
+    }
+}
+
 use self::Either::*;
 
 pub type EntidOr<T> = Either<Entid, T>;

db/src/schema.rs

@@ -223,30 +223,32 @@ impl SchemaBuilding for Schema {
 pub trait SchemaTypeChecking {
     /// Do schema-aware typechecking and coercion.
     ///
-    /// Either assert that the given value is in the attribute's value set, or (in limited cases)
+    /// Either assert that the given value is in the value type's value set, or (in limited cases)
-    /// coerce the given value into the attribute's value set.
+    /// coerce the given value into the value type's value set.
-    fn to_typed_value(&self, value: &edn::Value, attribute: &Attribute) -> Result<TypedValue>;
+    fn to_typed_value(&self, value: &edn::Value, value_type: ValueType) -> Result<TypedValue>;
 }
 
 impl SchemaTypeChecking for Schema {
-    fn to_typed_value(&self, value: &edn::Value, attribute: &Attribute) -> Result<TypedValue> {
+    fn to_typed_value(&self, value: &edn::Value, value_type: ValueType) -> Result<TypedValue> {
-        // TODO: encapsulate entid-ident-attribute for better error messages.
+        // TODO: encapsulate entid-ident-attribute for better error messages, perhaps by including
+        // the attribute (rather than just the attribute's value type) into this function or a
+        // wrapper function.
         match TypedValue::from_edn_value(value) {
             // We don't recognize this EDN at all.  Get out!
-            None => bail!(ErrorKind::BadEDNValuePair(value.clone(), attribute.value_type.clone())),
+            None => bail!(ErrorKind::BadEDNValuePair(value.clone(), value_type)),
-            Some(typed_value) => match (&attribute.value_type, typed_value) {
+            Some(typed_value) => match (value_type, typed_value) {
                 // Most types don't coerce at all.
-                (&ValueType::Boolean, tv @ TypedValue::Boolean(_)) => Ok(tv),
+                (ValueType::Boolean, tv @ TypedValue::Boolean(_)) => Ok(tv),
-                (&ValueType::Long, tv @ TypedValue::Long(_)) => Ok(tv),
+                (ValueType::Long, tv @ TypedValue::Long(_)) => Ok(tv),
-                (&ValueType::Double, tv @ TypedValue::Double(_)) => Ok(tv),
+                (ValueType::Double, tv @ TypedValue::Double(_)) => Ok(tv),
-                (&ValueType::String, tv @ TypedValue::String(_)) => Ok(tv),
+                (ValueType::String, tv @ TypedValue::String(_)) => Ok(tv),
-                (&ValueType::Uuid, tv @ TypedValue::Uuid(_)) => Ok(tv),
+                (ValueType::Uuid, tv @ TypedValue::Uuid(_)) => Ok(tv),
-                (&ValueType::Keyword, tv @ TypedValue::Keyword(_)) => Ok(tv),
+                (ValueType::Keyword, tv @ TypedValue::Keyword(_)) => Ok(tv),
                 // Ref coerces a little: we interpret some things depending on the schema as a Ref.
-                (&ValueType::Ref, TypedValue::Long(x)) => Ok(TypedValue::Ref(x)),
+                (ValueType::Ref, TypedValue::Long(x)) => Ok(TypedValue::Ref(x)),
-                (&ValueType::Ref, TypedValue::Keyword(ref x)) => self.require_entid(&x).map(|entid| TypedValue::Ref(entid)),
+                (ValueType::Ref, TypedValue::Keyword(ref x)) => self.require_entid(&x).map(|entid| TypedValue::Ref(entid)),
                 // Otherwise, we have a type mismatch.
-                (value_type, _) => bail!(ErrorKind::BadEDNValuePair(value.clone(), value_type.clone())),
+                (value_type, _) => bail!(ErrorKind::BadEDNValuePair(value.clone(), value_type)),
             }
         }
     }

db/src/tx.rs

@@ -51,6 +51,7 @@ use std::collections::{
     BTreeSet,
     VecDeque,
 };
+use std::rc::Rc;
 
 use db;
 use db::{
@@ -61,14 +62,16 @@ use entids;
 use errors::{ErrorKind, Result};
 use internal_types::{
     Either,
+    EntidOr,
     LookupRef,
     LookupRefOrTempId,
     TempIdHandle,
     TempIdMap,
     Term,
-    TermWithTempIdsAndLookupRefs,
     TermWithTempIds,
+    TermWithTempIdsAndLookupRefs,
     TermWithoutTempIds,
+    TypedValueOr,
     replace_lookup_ref};
 
 use mentat_core::{
@@ -191,23 +194,125 @@ impl<'conn, 'a> Tx<'conn, 'a> {
     /// The `Term` instances produce share interned TempId and LookupRef handles, and we return the
     /// interned handle sets so that consumers can ensure all handles are used appropriately.
     fn entities_into_terms_with_temp_ids_and_lookup_refs<I>(&self, entities: I) -> Result<(Vec<TermWithTempIdsAndLookupRefs>, intern_set::InternSet<TempId>, intern_set::InternSet<AVPair>)> where I: IntoIterator<Item=Entity> {
-        let mut temp_ids: intern_set::InternSet<TempId> = intern_set::InternSet::new();
+        struct InProcess<'a> {
-        let mut lookup_refs: intern_set::InternSet<AVPair> = intern_set::InternSet::new();
+            schema: &'a Schema,
+            mentat_id_count: i64,
+            temp_ids: intern_set::InternSet<TempId>,
+            lookup_refs: intern_set::InternSet<AVPair>,
+        }
 
-        let intern_lookup_ref = |lookup_refs: &mut intern_set::InternSet<AVPair>, lookup_ref: entmod::LookupRef| -> Result<LookupRef> {
+        impl<'a> InProcess<'a> {
-            let lr_a: i64 = match lookup_ref.a {
+            fn with_schema(schema: &'a Schema) -> InProcess<'a> {
-                entmod::Entid::Entid(ref a) => *a,
+                InProcess {
-                entmod::Entid::Ident(ref a) => self.schema.require_entid(&a)?,
+                    schema: schema,
-            };
+                    mentat_id_count: 0,
-            let lr_attribute: &Attribute = self.schema.require_attribute_for_entid(lr_a)?;
+                    temp_ids: intern_set::InternSet::new(),
-
+                    lookup_refs: intern_set::InternSet::new(),
-            if lr_attribute.unique.is_none() {
+                }
-                bail!(ErrorKind::NotYetImplemented(format!("Cannot resolve (lookup-ref {} {}) with attribute that is not :db/unique", lr_a, lookup_ref.v)))
             }
 
-            let lr_typed_value: TypedValue = self.schema.to_typed_value(&lookup_ref.v, &lr_attribute)?;
+            fn intern_lookup_ref(&mut self, lookup_ref: &entmod::LookupRef) -> Result<LookupRef> {
-            Ok(lookup_refs.intern((lr_a, lr_typed_value)))
+                let lr_a: i64 = match lookup_ref.a {
-        };
+                    entmod::Entid::Entid(ref a) => *a,
+                    entmod::Entid::Ident(ref a) => self.schema.require_entid(&a)?,
+                };
+                let lr_attribute: &Attribute = self.schema.require_attribute_for_entid(lr_a)?;
+
+                if lr_attribute.unique.is_none() {
+                    bail!(ErrorKind::NotYetImplemented(format!("Cannot resolve (lookup-ref {} {}) with attribute that is not :db/unique", lr_a, lookup_ref.v)))
+                }
+
+                let lr_typed_value: TypedValue = self.schema.to_typed_value(&lookup_ref.v, lr_attribute.value_type)?;
+                Ok(self.lookup_refs.intern((lr_a, lr_typed_value)))
+            }
+
+            fn intern_temp_id(&mut self, temp_id: TempId) -> Rc<TempId> {
+                self.temp_ids.intern(temp_id)
+            }
+
+            /// Allocate private internal tempids reserved for Mentat.  Internal tempids just need to be
+            /// unique within one transaction; they should never escape a transaction.
+            fn allocate_mentat_id(&mut self) -> entmod::EntidOrLookupRefOrTempId {
+                self.mentat_id_count += 1;
+                entmod::EntidOrLookupRefOrTempId::TempId(TempId::Internal(self.mentat_id_count))
+            }
+
+            fn entity_e_into_term_e(&mut self, x: entmod::EntidOrLookupRefOrTempId) -> Result<EntidOr<LookupRefOrTempId>> {
+                match x {
+                    entmod::EntidOrLookupRefOrTempId::Entid(e) => {
+                        let e: i64 = match e {
+                            entmod::Entid::Entid(ref e) => *e,
+                            entmod::Entid::Ident(ref e) => self.schema.require_entid(&e)?,
+                        };
+                        Ok(Either::Left(e))
+                    },
+
+                    entmod::EntidOrLookupRefOrTempId::TempId(e) => {
+                        Ok(Either::Right(LookupRefOrTempId::TempId(self.intern_temp_id(e))))
+                    },
+
+                    entmod::EntidOrLookupRefOrTempId::LookupRef(ref lookup_ref) => {
+                        Ok(Either::Right(LookupRefOrTempId::LookupRef(self.intern_lookup_ref(lookup_ref)?)))
+                    },
+                }
+            }
+
+            fn entity_a_into_term_a(&mut self, x: entmod::Entid) -> Result<Entid> {
+                let a = match x {
+                    entmod::Entid::Entid(ref a) => *a,
+                    entmod::Entid::Ident(ref a) => self.schema.require_entid(&a)?,
+                };
+                Ok(a)
+            }
+
+            fn entity_e_into_term_v(&mut self, x: entmod::EntidOrLookupRefOrTempId) -> Result<TypedValueOr<LookupRefOrTempId>> {
+                self.entity_e_into_term_e(x).map(|r| r.map_left(TypedValue::Ref))
+            }
+
+            fn entity_v_into_term_e(&mut self, x: entmod::AtomOrLookupRefOrVectorOrMapNotation, backward_a: &entmod::Entid) -> Result<EntidOr<LookupRefOrTempId>> {
+                match backward_a.unreversed() {
+                    None => {
+                        bail!(ErrorKind::NotYetImplemented(format!("Cannot explode map notation value in :attr/_reversed notation for forward attribute")));
+                    },
+                    Some(forward_a) => {
+                        let forward_a = self.entity_a_into_term_a(forward_a)?;
+                        let forward_attribute = self.schema.require_attribute_for_entid(forward_a)?;
+                        if forward_attribute.value_type != ValueType::Ref {
+                            bail!(ErrorKind::NotYetImplemented(format!("Cannot use :attr/_reversed notation for attribute {} that is not :db/valueType :db.type/ref", forward_a)))
+                        }
+
+                        match x {
+                            entmod::AtomOrLookupRefOrVectorOrMapNotation::Atom(ref v) => {
+                                // Here is where we do schema-aware typechecking: we either assert
+                                // that the given value is in the attribute's value set, or (in
+                                // limited cases) coerce the value into the attribute's value set.
+                                if let Some(text) = v.inner.as_text() {
+                                    Ok(Either::Right(LookupRefOrTempId::TempId(self.intern_temp_id(TempId::External(text.clone())))))
+                                } else {
+                                    if let TypedValue::Ref(entid) = self.schema.to_typed_value(&v.clone().without_spans(), ValueType::Ref)? {
+                                        Ok(Either::Left(entid))
+                                    } else {
+                                        // The given value is expected to be :db.type/ref, so this shouldn't happen.
+                                        bail!(ErrorKind::NotYetImplemented(format!("Cannot use :attr/_reversed notation for attribute {} with value that is not :db.valueType :db.type/ref", forward_a)))
+                                    }
+                                }
+                            },
+
+                            entmod::AtomOrLookupRefOrVectorOrMapNotation::LookupRef(ref lookup_ref) =>
+                                Ok(Either::Right(LookupRefOrTempId::LookupRef(self.intern_lookup_ref(lookup_ref)?))),
+
+                            entmod::AtomOrLookupRefOrVectorOrMapNotation::Vector(_) =>
+                                bail!(ErrorKind::NotYetImplemented(format!("Cannot explode vector value in :attr/_reversed notation for attribute {}", forward_a))),
+
+                            entmod::AtomOrLookupRefOrVectorOrMapNotation::MapNotation(_) =>
+                                bail!(ErrorKind::NotYetImplemented(format!("Cannot explode map notation value in :attr/_reversed notation for attribute {}", forward_a))),
+                        }
+                    },
+                }
+            }
+        }
+
+        let mut in_process = InProcess::with_schema(&self.schema);
 
         // We want to handle entities in the order they're given to us, while also "exploding" some
         // entities into many.  We therefore push the initial entities onto the back of the deque,
@@ -215,14 +320,6 @@ impl<'conn, 'a> Tx<'conn, 'a> {
         let mut deque: VecDeque<Entity> = VecDeque::default();
         deque.extend(entities);
 
-        // Allocate private internal tempids reserved for Mentat.  Internal tempids just need to be
-        // unique within one transaction; they should never escape a transaction.
-        let mut mentat_id_count = 0;
-        let mut allocate_mentat_id = move || {
-            mentat_id_count += 1;
-            entmod::EntidOrLookupRefOrTempId::TempId(TempId::Internal(mentat_id_count))
-        };
-
         let mut terms: Vec<TermWithTempIdsAndLookupRefs> = Vec::with_capacity(deque.len());
 
         while let Some(entity) = deque.pop_front() {
@@ -230,7 +327,7 @@ impl<'conn, 'a> Tx<'conn, 'a> {
                 Entity::MapNotation(mut map_notation) => {
                     // :db/id is optional; if it's not given, we generate a special internal tempid
                     // to use for upserting.  This tempid will not be reported in the TxReport.
-                    let db_id: entmod::EntidOrLookupRefOrTempId = mentat_tx_parser::remove_db_id(&mut map_notation)?.unwrap_or_else(&mut allocate_mentat_id);
+                    let db_id: entmod::EntidOrLookupRefOrTempId = mentat_tx_parser::remove_db_id(&mut map_notation)?.unwrap_or_else(|| in_process.allocate_mentat_id());
 
                     // We're not nested, so :db/isComponent is not relevant.  We just explode the
                     // map notation.
@@ -245,157 +342,135 @@ impl<'conn, 'a> Tx<'conn, 'a> {
                 },
 
                 Entity::AddOrRetract { op, e, a, v } => {
-                    let a: i64 = match a {
+                    if let Some(reversed_a) = a.unreversed() {
-                        entmod::Entid::Entid(ref a) => *a,
+                        let reversed_e = in_process.entity_v_into_term_e(v, &a)?;
-                        entmod::Entid::Ident(ref a) => self.schema.require_entid(&a)?,
+                        let reversed_a = in_process.entity_a_into_term_a(reversed_a)?;
-                    };
+                        let reversed_v = in_process.entity_e_into_term_v(e)?;
+                        terms.push(Term::AddOrRetract(OpType::Add, reversed_e, reversed_a, reversed_v));
+                    } else {
+                        let a = in_process.entity_a_into_term_a(a)?;
+                        let attribute = self.schema.require_attribute_for_entid(a)?;
 
-                    let attribute: &Attribute = self.schema.require_attribute_for_entid(a)?;
+                        let v = match v {
+                            entmod::AtomOrLookupRefOrVectorOrMapNotation::Atom(v) => {
+                                if attribute.value_type == ValueType::Ref && v.inner.is_text() {
+                                    Either::Right(LookupRefOrTempId::TempId(in_process.intern_temp_id(v.inner.as_text().cloned().map(TempId::External).unwrap())))
+                                } else {
+                                    // Here is where we do schema-aware typechecking: we either assert that
+                                    // the given value is in the attribute's value set, or (in limited
+                                    // cases) coerce the value into the attribute's value set.
+                                    let typed_value: TypedValue = self.schema.to_typed_value(&v.without_spans(), attribute.value_type)?;
+                                    Either::Left(typed_value)
+                                }
+                            },
 
-                    let v = match v {
+                            entmod::AtomOrLookupRefOrVectorOrMapNotation::LookupRef(ref lookup_ref) => {
-                        entmod::AtomOrLookupRefOrVectorOrMapNotation::Atom(v) => {
+                                if attribute.value_type != ValueType::Ref {
-                            if attribute.value_type == ValueType::Ref && v.inner.is_text() {
+                                    bail!(ErrorKind::NotYetImplemented(format!("Cannot resolve value lookup ref for attribute {} that is not :db/valueType :db.type/ref", a)))
-                                Either::Right(LookupRefOrTempId::TempId(temp_ids.intern(v.inner.as_text().cloned().map(TempId::External).unwrap())))
+                                }
-                            } else {
-                                // Here is where we do schema-aware typechecking: we either assert that
-                                // the given value is in the attribute's value set, or (in limited
-                                // cases) coerce the value into the attribute's value set.
-                                let typed_value: TypedValue = self.schema.to_typed_value(&v.without_spans(), &attribute)?;
-                                Either::Left(typed_value)
-                            }
-                        },
 
-                        entmod::AtomOrLookupRefOrVectorOrMapNotation::LookupRef(lookup_ref) => {
+                                Either::Right(LookupRefOrTempId::LookupRef(in_process.intern_lookup_ref(lookup_ref)?))
-                            if attribute.value_type != ValueType::Ref {
+                            },
-                                bail!(ErrorKind::NotYetImplemented(format!("Cannot resolve value lookup ref for attribute {} that is not :db/valueType :db.type/ref", a)))
-                            }
 
-                            Either::Right(LookupRefOrTempId::LookupRef(intern_lookup_ref(&mut lookup_refs, lookup_ref)?))
+                            entmod::AtomOrLookupRefOrVectorOrMapNotation::Vector(vs) => {
-                        },
+                                if !attribute.multival {
+                                    bail!(ErrorKind::NotYetImplemented(format!("Cannot explode vector value for attribute {} that is not :db.cardinality :db.cardinality/many", a)));
+                                }
 
-                        entmod::AtomOrLookupRefOrVectorOrMapNotation::Vector(vs) => {
+                                for vv in vs {
-                            if !attribute.multival {
+                                    deque.push_front(Entity::AddOrRetract {
-                                bail!(ErrorKind::NotYetImplemented(format!("Cannot explode vector value for attribute {} that is not :db.cardinality :db.cardinality/many", a)));
+                                        op: op.clone(),
-                            }
+                                        e: e.clone(),
+                                        a: entmod::Entid::Entid(a),
+                                        v: vv,
+                                    });
+                                }
+                                continue
+                            },
 
-                            for vv in vs {
+                            entmod::AtomOrLookupRefOrVectorOrMapNotation::MapNotation(mut map_notation) => {
-                                deque.push_front(Entity::AddOrRetract {
+                                // TODO: consider handling this at the tx-parser level.  That would be
-                                    op: op.clone(),
+                                // more strict and expressive, but it would lead to splitting
-                                    e: e.clone(),
+                                // AddOrRetract, which proliferates types and code, or only handling
-                                    a: entmod::Entid::Entid(a),
+                                // nested maps rather than map values, like Datomic does.
-                                    v: vv,
+                                if op != OpType::Add {
-                                });
+                                    bail!(ErrorKind::NotYetImplemented(format!("Cannot explode nested map value in :db/retract for attribute {}", a)));
-                            }
+                                }
-                            continue
-                        },
 
-                        entmod::AtomOrLookupRefOrVectorOrMapNotation::MapNotation(mut map_notation) => {
+                                if attribute.value_type != ValueType::Ref {
-                            // TODO: consider handling this at the tx-parser level.  That would be
+                                    bail!(ErrorKind::NotYetImplemented(format!("Cannot explode nested map value for attribute {} that is not :db/valueType :db.type/ref", a)))
-                            // more strict and expressive, but it would lead to splitting
+                                }
-                            // AddOrRetract, which proliferates types and code, or only handling
-                            // nested maps rather than map values, like Datomic does.
-                            if op != OpType::Add {
-                                bail!(ErrorKind::NotYetImplemented(format!("Cannot explode nested map value in :db/retract for attribute {}", a)));
-                            }
 
-                            if attribute.value_type != ValueType::Ref {
+                                // :db/id is optional; if it's not given, we generate a special internal tempid
-                                bail!(ErrorKind::NotYetImplemented(format!("Cannot explode nested map value for attribute {} that is not :db/valueType :db.type/ref", a)))
+                                // to use for upserting.  This tempid will not be reported in the TxReport.
-                            }
+                                let db_id: Option<entmod::EntidOrLookupRefOrTempId> = mentat_tx_parser::remove_db_id(&mut map_notation)?;
+                                let mut dangling = db_id.is_none();
+                                let db_id: entmod::EntidOrLookupRefOrTempId = db_id.unwrap_or_else(|| in_process.allocate_mentat_id());
 
-                            // :db/id is optional; if it's not given, we generate a special internal tempid
+                                // We're nested, so we want to ensure we're not creating "dangling"
-                            // to use for upserting.  This tempid will not be reported in the TxReport.
+                                // entities that can't be reached.  If we're :db/isComponent, then this
-                            let db_id: Option<entmod::EntidOrLookupRefOrTempId> = mentat_tx_parser::remove_db_id(&mut map_notation)?;
+                                // is not dangling.  Otherwise, the resulting map needs to have a
-                            let mut dangling = db_id.is_none();
+                                // :db/unique :db.unique/identity [a v] pair, so that it's reachable.
-                            let db_id: entmod::EntidOrLookupRefOrTempId = db_id.unwrap_or_else(&mut allocate_mentat_id);
+                                // Per http://docs.datomic.com/transactions.html: "Either the reference
-
+                                // to the nested map must be a component attribute, or the nested map
-                            // We're nested, so we want to ensure we're not creating "dangling"
+                                // must include a unique attribute. This constraint prevents the
-                            // entities that can't be reached.  If we're :db/isComponent, then this
+                                // accidental creation of easily-orphaned entities that have no identity
-                            // is not dangling.  Otherwise, the resulting map needs to have a
+                                // or relation to other entities."
-                            // :db/unique :db.unique/identity [a v] pair, so that it's reachable.
+                                if attribute.component {
-                            // Per http://docs.datomic.com/transactions.html: "Either the reference
-                            // to the nested map must be a component attribute, or the nested map
-                            // must include a unique attribute. This constraint prevents the
-                            // accidental creation of easily-orphaned entities that have no identity
-                            // or relation to other entities."
-                            if attribute.component {
-                                dangling = false;
-                            }
-
-                            for (inner_a, inner_v) in map_notation {
-                                let inner_entid: i64 = match inner_a {
-                                    entmod::Entid::Entid(ref a) => *a,
-                                    entmod::Entid::Ident(ref a) => self.schema.require_entid(&a)?,
-                                };
-
-                                let inner_attribute: &Attribute = self.schema.require_attribute_for_entid(inner_entid)?;
-                                if inner_attribute.unique == Some(attribute::Unique::Identity) {
                                     dangling = false;
                                 }
 
-                                deque.push_front(Entity::AddOrRetract {
+                                for (inner_a, inner_v) in map_notation {
-                                    op: OpType::Add,
+                                    if let Some(reversed_a) = inner_a.unreversed() {
-                                    e: db_id.clone(),
+                                        // We definitely have a reference.  The reference might be
-                                    a: entmod::Entid::Entid(inner_entid),
+                                        // dangling (a bare entid, for example), but we don't yet
-                                    v: inner_v,
+                                        // support nested maps and reverse notation simultaneously
-                                });
+                                        // (i.e., we don't accept {:reverse/_attribute {:nested map}})
-                            }
+                                        // so we don't need to check that the nested map reference isn't
+                                        // dangling.
+                                        dangling = false;
 
-                            if dangling {
+                                        let reversed_e = in_process.entity_v_into_term_e(inner_v, &inner_a)?;
-                                bail!(ErrorKind::NotYetImplemented(format!("Cannot explode nested map value that would lead to dangling entity for attribute {}", a)));
+                                        let reversed_a = in_process.entity_a_into_term_a(reversed_a)?;
-                            }
+                                        let reversed_v = in_process.entity_e_into_term_v(db_id.clone())?;
+                                        terms.push(Term::AddOrRetract(OpType::Add, reversed_e, reversed_a, reversed_v));
+                                    } else {
+                                        let inner_a = in_process.entity_a_into_term_a(inner_a)?;
+                                        let inner_attribute = self.schema.require_attribute_for_entid(inner_a)?;
+                                        if inner_attribute.unique == Some(attribute::Unique::Identity) {
+                                            dangling = false;
+                                        }
 
-                            // Similar, but not identical, to the expansion of the entity position e
+                                        deque.push_front(Entity::AddOrRetract {
-                            // below.  This returns Either::Left(TypedValue) instances; that returns
+                                            op: OpType::Add,
-                            // Either::Left(i64) instances.
+                                            e: db_id.clone(),
-                            match db_id {
+                                            a: entmod::Entid::Entid(inner_a),
-                                entmod::EntidOrLookupRefOrTempId::Entid(e) => {
+                                            v: inner_v,
-                                    let e: i64 = match e {
+                                        });
-                                        entmod::Entid::Entid(ref e) => *e,
+                                    }
-                                        entmod::Entid::Ident(ref e) => self.schema.require_entid(&e)?,
+                                }
-                                    };
-                                    Either::Left(TypedValue::Ref(e))
-                                },
 
-                                entmod::EntidOrLookupRefOrTempId::TempId(e) => {
+                                if dangling {
-                                    Either::Right(LookupRefOrTempId::TempId(temp_ids.intern(e)))
+                                    bail!(ErrorKind::NotYetImplemented(format!("Cannot explode nested map value that would lead to dangling entity for attribute {}", a)));
-                                },
+                                }
 
-                                entmod::EntidOrLookupRefOrTempId::LookupRef(lookup_ref) => {
+                                in_process.entity_e_into_term_v(db_id)?
-                                    Either::Right(LookupRefOrTempId::LookupRef(intern_lookup_ref(&mut lookup_refs, lookup_ref)?))
+                            },
-                                },
+                        };
-                            }
-                        },
-                    };
 
-                    let e = match e {
+                        let e = in_process.entity_e_into_term_e(e)?;
-                        entmod::EntidOrLookupRefOrTempId::Entid(e) => {
+                        terms.push(Term::AddOrRetract(op, e, a, v));
-                            let e: i64 = match e {
+                    }
-                                entmod::Entid::Entid(ref e) => *e,
-                                entmod::Entid::Ident(ref e) => self.schema.require_entid(&e)?,
-                            };
-                            Either::Left(e)
-                        },
-
-                        entmod::EntidOrLookupRefOrTempId::TempId(e) => {
-                            Either::Right(LookupRefOrTempId::TempId(temp_ids.intern(e)))
-                        },
-
-                        entmod::EntidOrLookupRefOrTempId::LookupRef(lookup_ref) => {
-                            Either::Right(LookupRefOrTempId::LookupRef(intern_lookup_ref(&mut lookup_refs, lookup_ref)?))
-                        },
-                    };
-
-                    terms.push(Term::AddOrRetract(op, e, a, v));
                 },
             }
         };
-        Ok((terms, temp_ids, lookup_refs))
+        Ok((terms, in_process.temp_ids, in_process.lookup_refs))
     }
 
     /// Pipeline stage 2: rewrite `Term` instances with lookup refs into `Term` instances without
     /// lookup refs.
     ///
-    /// The `Term` instances produce share interned TempId handles and have no LookupRef references.
+    /// The `Term` instances produced share interned TempId handles and have no LookupRef references.
     fn resolve_lookup_refs<I>(&self, lookup_ref_map: &AVMap, terms: I) -> Result<Vec<TermWithTempIds>> where I: IntoIterator<Item=TermWithTempIdsAndLookupRefs> {
         terms.into_iter().map(|term: TermWithTempIdsAndLookupRefs| -> Result<TermWithTempIds> {
             match term {

edn/src/symbols.rs

@@ -214,6 +214,30 @@ impl NamespacedKeyword {
             namespace: self.namespace.clone(),
         }
     }
+
+    /// If this `NamespacedKeyword` is 'backward' (see `symbols::NamespacedKeyword::is_backward`),
+    /// return `Some('forward name')`; otherwise, return `None`.
+    ///
+    /// # Examples
+    ///
+    /// ```rust
+    /// # use edn::symbols::NamespacedKeyword;
+    /// let nsk = NamespacedKeyword::new("foo", "bar");
+    /// assert_eq!(None, nsk.unreversed());
+    ///
+    /// let reversed = nsk.to_reversed();
+    /// assert_eq!(Some(nsk), reversed.unreversed());
+    /// ```
+    pub fn unreversed(&self) -> Option<NamespacedKeyword> {
+        if self.is_backward() {
+            Some(NamespacedKeyword {
+                name: self.name[1..].to_string(),
+                namespace: self.namespace.clone(),
+            })
+        } else {
+            None
+        }
+    }
 }
 
 //

edn/src/types.rs

@@ -594,7 +594,6 @@ mod test {
 
     use chrono::{
         DateTime,
-        TimeZone,
         UTC,
     };
     use num::BigInt;

parser-utils/src/value_and_span.rs

@@ -452,6 +452,26 @@ pub fn namespaced_keyword<'a>() -> Expected<FnParser<Stream<'a>, fn(Stream<'a>)
     parser(namespaced_keyword_ as fn(Stream<'a>) -> ParseResult<&'a edn::NamespacedKeyword, Stream<'a>>).expected("namespaced_keyword")
 }
 
+pub fn forward_keyword_<'a>(input: Stream<'a>) -> ParseResult<&'a edn::NamespacedKeyword, Stream<'a>> {
+    satisfy_map(|v: &'a edn::ValueAndSpan| v.inner.as_namespaced_keyword().and_then(|k| if k.is_forward() { Some(k) } else { None }))
+        .parse_lazy(input)
+        .into()
+}
+
+pub fn forward_keyword<'a>() -> Expected<FnParser<Stream<'a>, fn(Stream<'a>) -> ParseResult<&'a edn::NamespacedKeyword, Stream<'a>>>> {
+    parser(forward_keyword_ as fn(Stream<'a>) -> ParseResult<&'a edn::NamespacedKeyword, Stream<'a>>).expected("forward_keyword")
+}
+
+pub fn backward_keyword_<'a>(input: Stream<'a>) -> ParseResult<&'a edn::NamespacedKeyword, Stream<'a>> {
+    satisfy_map(|v: &'a edn::ValueAndSpan| v.inner.as_namespaced_keyword().and_then(|k| if k.is_backward() { Some(k) } else { None }))
+        .parse_lazy(input)
+        .into()
+}
+
+pub fn backward_keyword<'a>() -> Expected<FnParser<Stream<'a>, fn(Stream<'a>) -> ParseResult<&'a edn::NamespacedKeyword, Stream<'a>>>> {
+    parser(backward_keyword_ as fn(Stream<'a>) -> ParseResult<&'a edn::NamespacedKeyword, Stream<'a>>).expected("backward_keyword")
+}
+
 /// Generate a `satisfy` expression that matches a `PlainSymbol` value with the given name.
 ///
 /// We do this rather than using `combine::token` so that we don't need to allocate a new `String`

tx-parser/src/lib.rs

@@ -45,6 +45,8 @@ use mentat_parser_utils::{ResultParser};
 use mentat_parser_utils::value_and_span::{
     Item,
     OfExactlyParsing,
+    backward_keyword,
+    forward_keyword,
     integer,
     list,
     map,
@@ -57,12 +59,25 @@ pub use errors::*;
 
 pub struct Tx<'a>(std::marker::PhantomData<&'a ()>);
 
+// Accepts entid, :attribute/forward, and :attribute/_backward.
 def_parser!(Tx, entid, Entid, {
     integer()
         .map(|x| Entid::Entid(x))
         .or(namespaced_keyword().map(|x| Entid::Ident(x.clone())))
 });
 
+// Accepts entid and :attribute/forward.
+def_parser!(Tx, forward_entid, Entid, {
+    integer()
+        .map(|x| Entid::Entid(x))
+        .or(forward_keyword().map(|x| Entid::Ident(x.clone())))
+});
+
+// Accepts only :attribute/_backward.
+def_parser!(Tx, backward_entid, Entid, {
+    backward_keyword().map(|x| Entid::Ident(x.to_reversed()))
+});
+
 def_matches_plain_symbol!(Tx, literal_lookup_ref, "lookup-ref");
 
 def_parser!(Tx, lookup_ref, LookupRef, {
@@ -106,18 +121,23 @@ def_matches_namespaced_keyword!(Tx, literal_db_retract, "db", "retract");
 
 def_parser!(Tx, add_or_retract, Entity, {
     vector().of_exactly(
+        // TODO: This commits as soon as it sees :db/{add,retract}, but could use an improved error message.
         (Tx::literal_db_add().map(|_| OpType::Add).or(Tx::literal_db_retract().map(|_| OpType::Retract)),
-             Tx::entid_or_lookup_ref_or_temp_id(),
+          try((Tx::entid_or_lookup_ref_or_temp_id(),
-             Tx::entid(),
+               Tx::forward_entid(),
-             Tx::atom_or_lookup_ref_or_vector())
+               Tx::atom_or_lookup_ref_or_vector()))
-        .map(|(op, e, a, v)| {
+          .or(try((Tx::atom_or_lookup_ref_or_vector(),
-            Entity::AddOrRetract {
+                   Tx::backward_entid(),
-                op: op,
+                   Tx::entid_or_lookup_ref_or_temp_id()))
-                e: e,
+              .map(|(v, a, e)| (e, a, v))))
-                a: a,
+            .map(|(op, (e, a, v))| {
-                v: v,
+                Entity::AddOrRetract {
-            }
+                    op: op,
-        }))
+                    e: e,
+                    a: a,
+                    v: v,
+                }
+            }))
 });
 
 def_parser!(Tx, map_notation, MapNotation, {

tx-parser/tests/parser.rs

@@ -85,4 +85,22 @@ fn test_entities() {
                ]);
 }
 
+#[test]
+fn test_reverse_notation_illegal_nested_values() {
+    // Verify that we refuse to parse direct reverse notation with nested value maps or vectors.
+
+    let input = "[[:db/add 100 :test/_dangling {:test/many 13}]]";
+    let edn = parse::value(input).expect("to parse test input");
+    let result = Tx::parse(&edn);
+    // TODO: it would be much better to assert details about the error (here and below), but right
+    // now the error message isn't clear that the given value isn't valid for the backward attribute
+    // :test/_dangling.
+    assert!(result.is_err());
+
+    let input = "[[:db/add 100 :test/_dangling [:test/many 13]]]";
+    let edn = parse::value(input).expect("to parse test input");
+    let result = Tx::parse(&edn);
+    assert!(result.is_err());
+}
+
 // TODO: test error handling in select cases.

tx/src/entities.rs

@@ -56,6 +56,29 @@ pub enum Entid {
     Ident(NamespacedKeyword),
 }
 
+impl Entid {
+    pub fn is_backward(&self) -> bool {
+        match self {
+            &Entid::Entid(_) => false,
+            &Entid::Ident(ref a) => a.is_backward(),
+        }
+    }
+
+    pub fn to_reversed(&self) -> Option<Entid> {
+        match self {
+            &Entid::Entid(_) => None,
+            &Entid::Ident(ref a) => Some(Entid::Ident(a.to_reversed())),
+        }
+    }
+
+    pub fn unreversed(&self) -> Option<Entid> {
+        match self {
+            &Entid::Entid(_) => None,
+            &Entid::Ident(ref a) => a.unreversed().map(Entid::Ident),
+        }
+    }
+}
+
 #[derive(Clone, Debug, Eq, Hash, Ord, PartialOrd, PartialEq)]
 pub struct LookupRef {
     pub a: Entid,