- mutable primary key support is added. primary key columns can be

changed freely, and the identity of the instance will change upon
flush. In addition, update cascades of foreign key referents (primary
key or not) along relations are supported, either in tandem with the
database's ON UPDATE CASCADE (required for DB's like Postgres) or
issued directly by the ORM in the form of UPDATE statements, by setting
the flag "passive_cascades=False".

13 files changed, 638 insertions, 98 deletions

diff --git a/CHANGES b/CHANGES
index c9fa81e29..ae9409498 100644
--- a/CHANGES
+++ b/CHANGES
@@ -59,6 +59,14 @@ CHANGES
      have not, you might notice your apps using a lot fewer queries than
      before in some situations. [ticket:871]
 
+   - mutable primary key support is added. primary key columns can be
+     changed freely, and the identity of the instance will change upon
+     flush. In addition, update cascades of foreign key referents (primary
+     key or not) along relations are supported, either in tandem with the
+     database's ON UPDATE CASCADE (required for DB's like Postgres) or
+     issued directly by the ORM in the form of UPDATE statements, by setting
+     the flag "passive_cascades=False".
+     
    - added "cascade delete" behavior to "dynamic" relations just like
      that of regular relations.  if passive_deletes flag (also just added)
      is not set, a delete of the parent item will trigger a full load of 
diff --git a/doc/build/content/mappers.txt b/doc/build/content/mappers.txt
index 6d98afdb9..fdf16e920 100644
--- a/doc/build/content/mappers.txt
+++ b/doc/build/content/mappers.txt
@@ -1385,4 +1385,30 @@ Use `passive_deletes=True` to disable child object loading on a DELETE operation
 
 When `passive_deletes` is applied, the `children` relation will not be loaded into memory when an instance of `MyClass` is marked for deletion.  The `cascade="all, delete-orphan"` *will* take effect for instances of `MyOtherClass` which are currently present in the session; however for instances of `MyOtherClass` which are not loaded, SQLAlchemy assumes that "ON DELETE CASCADE" rules will ensure that those rows are deleted by the database and that no foreign key violation will occur.
         
+#### Mutable Primary Keys / Update Cascades {@name=mutablepks}
+
+As of SQLAlchemy 0.4.2, the primary key attributes of an instance can be changed freely, and will be persisted upon flush.  When the primary key of an entity changes, related items which reference the primary key must also be updated as well.  For databases which enforce referential integrity, it's required to use the database's ON UPDATE CASCADE functionality in order to propagate primary key changes.  For those which don't, the `passive_cascades` flag can be set to `False` which instructs SQLAlchemy to issue UPDATE statements individually.  The `passive_cascades` flag can also be `False` in conjunction with ON UPDATE CASCADE functionality, although in that case it issues UPDATE statements unnecessarily.
+
+A typical mutable primary key setup might look like:
+
+    {python}
+    users = Table('users', metadata,
+        Column('username', String(50), primary_key=True),
+        Column('fullname', String(100)))
+        
+    addresses = Table('addresses', metadata,
+        Column('email', String(50), primary_key=True),
+        Column('username', String(50), ForeignKey('users.username', onupdate="cascade")))
+    
+    class User(object):
+        pass
+    class Address(object):
+        pass
+
+    mapper(User, users, properties={
+        'addresses':relation(Address, passive_updates=False)
+    })
+    mapper(Address, addresses)
+    
+passive_updates is set to `True` by default.  Foreign key references to non-primary key columns are supported as well.
 
diff --git a/lib/sqlalchemy/orm/__init__.py b/lib/sqlalchemy/orm/__init__.py
index 7ce298f71..56edc03f2 100644
--- a/lib/sqlalchemy/orm/__init__.py
+++ b/lib/sqlalchemy/orm/__init__.py
@@ -170,7 +170,8 @@ def relation(argument, secondary=None, **kwargs):
           indicates the ordering that should be applied when loading these items.
 
         passive_deletes=False
-          Indicates the behavior of delete operations. 
+          Indicates loading behavior during delete operations. 
+          
           A value of True indicates that unloaded child items should not be loaded
           during a delete operation on the parent.  Normally, when a parent
           item is deleted, all child items are loaded so that they can either be
@@ -185,7 +186,29 @@ def relation(argument, secondary=None, **kwargs):
           or error raise scenario is in place on the database side.  Note that
           the foreign key attributes on in-session child objects will not be changed
           after a flush occurs so this is a very special use-case setting.
-
+         
+        passive_updates=True
+          Indicates loading and INSERT/UPDATE/DELETE behavior when the source
+          of a foreign key value changes (i.e. an "on update" cascade), which are
+          typically the primary key columns of the source row.
+          
+          When True, it is assumed that ON UPDATE CASCADE is configured on the
+          foreign key in the database, and that the database will handle propagation of an
+          UPDATE from a source column to dependent rows.  Note that with databases
+          which enforce referential integrity (ie. Postgres, MySQL with InnoDB tables),
+          ON UPDATE CASCADE is required for this operation.  The relation() will 
+          update the value of the attribute on related items which are locally present
+          in the session during a flush.
+          
+          When False, it is assumed that the database does not enforce referential
+          integrity and will not be issuing its own CASCADE operation for an update.
+          The relation() will issue the appropriate UPDATE statements to the database
+          in response to the change of a referenced key, and items locally present
+          in the session during a flush will also be refreshed.  
+          
+          This flag should probably be set to False if primary key changes are expected
+          and the database in use doesn't support CASCADE (i.e. SQLite, MySQL MyISAM tables).
+          
         post_update
           this indicates that the relationship should be handled by a second
           UPDATE statement after an INSERT or before a DELETE. Currently, it also
diff --git a/lib/sqlalchemy/orm/attributes.py b/lib/sqlalchemy/orm/attributes.py
index b73ec0e00..bc9da18ba 100644
--- a/lib/sqlalchemy/orm/attributes.py
+++ b/lib/sqlalchemy/orm/attributes.py
@@ -259,7 +259,15 @@ class AttributeImpl(object):
 
     def set(self, state, value, initiator):
         raise NotImplementedError()
-
+    
+    def get_committed_value(self, state):
+        if state.committed_state is not None:
+            if self.key not in state.committed_state:
+                self.get()
+            return state.committed_state.get(self.key)
+        else:
+            return None
+            
     def set_committed_value(self, state, value):
         """set an attribute value on the given instance and 'commit' it.
         
diff --git a/lib/sqlalchemy/orm/dependency.py b/lib/sqlalchemy/orm/dependency.py
index ae499ce1e..0a097fd24 100644
--- a/lib/sqlalchemy/orm/dependency.py
+++ b/lib/sqlalchemy/orm/dependency.py
@@ -27,6 +27,8 @@ def create_dependency_processor(prop):
         return types[prop.direction](prop)
 
 class DependencyProcessor(object):
+    no_dependencies = False
+    
     def __init__(self, prop):
         self.prop = prop
         self.cascade = prop.cascade
@@ -39,6 +41,7 @@ class DependencyProcessor(object):
         self.post_update = prop.post_update
         self.foreign_keys = prop.foreign_keys
         self.passive_deletes = prop.passive_deletes
+        self.passive_updates = prop.passive_updates
         self.enable_typechecks = prop.enable_typechecks
         self.key = prop.key
 
@@ -133,26 +136,6 @@ class DependencyProcessor(object):
         else:
             self.syncrules.compile(self.prop.primaryjoin, foreign_keys=self.foreign_keys)
 
-    def get_object_dependencies(self, state, uowcommit, passive = True):
-        key = ("dependencies", state, self.key, passive)
-        
-        # cache the objects, not the states; the strong reference here
-        # prevents newly loaded objects from being dereferenced during the 
-        # flush process
-        if key in uowcommit.attributes:
-            (added, unchanged, deleted) = uowcommit.attributes[key]
-        else:
-            (added, unchanged, deleted) = attributes.get_history(state, self.key, passive = passive)
-            uowcommit.attributes[key] = (added, unchanged, deleted)
-            
-        if added is None:
-            return (added, unchanged, deleted)
-        else:
-            return (
-                [getattr(c, '_state', None) for c in added],
-                [getattr(c, '_state', None) for c in unchanged],
-                [getattr(c, '_state', None) for c in deleted],
-                )
 
     def _conditional_post_update(self, state, uowcommit, related):
         """Execute a post_update call.
@@ -173,7 +156,10 @@ class DependencyProcessor(object):
                 if x is not None:
                     uowcommit.register_object(state, postupdate=True, post_update_cols=self.syncrules.dest_columns())
                     break
-
+    
+    def _pks_changed(self, uowcommit, state):
+        return self.syncrules.source_changes(uowcommit, state)
+        
     def __str__(self):
         return "%s(%s)" % (self.__class__.__name__, str(self.prop))
 
@@ -198,7 +184,7 @@ class OneToManyDP(DependencyProcessor):
             # is on.
             if (not self.cascade.delete or self.post_update) and not self.passive_deletes=='all':
                 for state in deplist:
-                    (added, unchanged, deleted) = self.get_object_dependencies(state, uowcommit, passive=self.passive_deletes)
+                    (added, unchanged, deleted) = uowcommit.get_attribute_history(state, self.key,passive=self.passive_deletes)
                     if unchanged or deleted:
                         for child in deleted:
                             if child is not None and self.hasparent(child) is False:
@@ -210,7 +196,7 @@ class OneToManyDP(DependencyProcessor):
                                 self._conditional_post_update(child, uowcommit, [state])
         else:
             for state in deplist:
-                (added, unchanged, deleted) = self.get_object_dependencies(state, uowcommit, passive=True)
+                (added, unchanged, deleted) = uowcommit.get_attribute_history(state, self.key, passive=True)
                 if added or deleted:
                     for child in added:
                         self._synchronize(state, child, None, False, uowcommit)
@@ -219,7 +205,12 @@ class OneToManyDP(DependencyProcessor):
                     for child in deleted:
                         if not self.cascade.delete_orphan and not self.hasparent(child):
                             self._synchronize(state, child, None, True, uowcommit)
-
+                            
+                if self._pks_changed(uowcommit, state):
+                    if unchanged:
+                        for child in unchanged:
+                            self._synchronize(state, child, None, False, uowcommit)
+                
     def preprocess_dependencies(self, task, deplist, uowcommit, delete = False):
         #print self.mapper.mapped_table.name + " " + self.key + " " + repr(len(deplist)) + " preprocess_dep isdelete " + repr(delete) + " direction " + repr(self.direction)
 
@@ -228,7 +219,7 @@ class OneToManyDP(DependencyProcessor):
             # the child objects have to have their foreign key to the parent set to NULL
             if not self.post_update and not self.cascade.delete and not self.passive_deletes=='all':
                 for state in deplist:
-                    (added, unchanged, deleted) = self.get_object_dependencies(state, uowcommit, passive=self.passive_deletes)
+                    (added, unchanged, deleted) = uowcommit.get_attribute_history(state, self.key,passive=self.passive_deletes)
                     if unchanged or deleted:
                         for child in deleted:
                             if child is not None and self.hasparent(child) is False:
@@ -238,7 +229,7 @@ class OneToManyDP(DependencyProcessor):
                                 uowcommit.register_object(child)
         else:
             for state in deplist:
-                (added, unchanged, deleted) = self.get_object_dependencies(state, uowcommit, passive=True)
+                (added, unchanged, deleted) = uowcommit.get_attribute_history(state, self.key,passive=True)
                 if added or deleted:
                     for child in added:
                         if child is not None:
@@ -250,7 +241,13 @@ class OneToManyDP(DependencyProcessor):
                             uowcommit.register_object(child, isdelete=True)
                             for c, m in self.mapper.cascade_iterator('delete', child):
                                 uowcommit.register_object(c._state, isdelete=True)
-
+                if not self.passive_updates and self._pks_changed(uowcommit, state):
+                    if not unchanged:
+                        (added, unchanged, deleted) = uowcommit.get_attribute_history(state, self.key, passive=False)
+                    if unchanged:
+                        for child in unchanged:
+                            uowcommit.register_object(child)
+                    
     def _synchronize(self, state, child, associationrow, clearkeys, uowcommit):
         if child is not None:
             child = getattr(child, '_state', child)
@@ -261,7 +258,50 @@ class OneToManyDP(DependencyProcessor):
         self._verify_canload(child)
         self.syncrules.execute(source, dest, source, child, clearkeys)
 
+class DetectKeySwitch(DependencyProcessor):
+    """a special DP that works for many-to-one relations, fires off for
+    child items who have changed their referenced key."""
+
+    no_dependencies = True
+    
+    def register_dependencies(self, uowcommit):
+        uowcommit.register_processor(self.parent, self, self.mapper)
+        
+    def preprocess_dependencies(self, task, deplist, uowcommit, delete=False):
+        # for non-passive updates, register in the preprocess stage
+        # so that mapper save_obj() gets a hold of changes
+        if not delete and not self.passive_updates:
+            self._process_key_switches(deplist, uowcommit)
+        
+    def process_dependencies(self, task, deplist, uowcommit, delete=False):
+        # for passive updates, register objects in the process stage
+        # so that we avoid ManyToOneDP's registering the object without 
+        # the listonly flag in its own preprocess stage (results in UPDATE)
+        # statements being emitted
+        if not delete and self.passive_updates:
+            self._process_key_switches(deplist, uowcommit)
+    
+    def _process_key_switches(self, deplist, uowcommit):    
+        switchers = util.Set(s for s in deplist if self._pks_changed(uowcommit, s))
+        if switchers:
+            # yes, we're doing a linear search right now through the UOW.  only 
+            # takes effect when primary key values have actually changed.
+            # a possible optimization might be to enhance the "hasparents" capability of
+            # attributes to actually store all parent references, but this introduces
+            # more complicated attribute accounting.
+            for s in [elem for elem in uowcommit.session.identity_map.all_states() 
+                if issubclass(elem.class_, self.parent.class_) and 
+                    self.key in elem.dict and 
+                    elem.dict[self.key]._state in switchers
+                ]:
+                uowcommit.register_object(s, listonly=self.passive_updates)
+                self.syncrules.execute(s.dict[self.key]._state, s, None, None, False)
+                        
 class ManyToOneDP(DependencyProcessor):
+    def __init__(self, prop):
+        DependencyProcessor.__init__(self, prop)
+        self.mapper._dependency_processors.append(DetectKeySwitch(prop))
+        
     def register_dependencies(self, uowcommit):
         if self.post_update:
             if not self.is_backref:
@@ -272,6 +312,7 @@ class ManyToOneDP(DependencyProcessor):
         else:
             uowcommit.register_dependency(self.mapper, self.parent)
             uowcommit.register_processor(self.mapper, self, self.parent)
+            
 
     def process_dependencies(self, task, deplist, uowcommit, delete = False):
         #print self.mapper.mapped_table.name + " " + self.key + " " + repr(len(deplist)) + " process_dep isdelete " + repr(delete) + " direction " + repr(self.direction)
@@ -281,12 +322,12 @@ class ManyToOneDP(DependencyProcessor):
                 # before we can DELETE the row
                 for state in deplist:
                     self._synchronize(state, None, None, True, uowcommit)
-                    (added, unchanged, deleted) = self.get_object_dependencies(state, uowcommit, passive=self.passive_deletes)
+                    (added, unchanged, deleted) = uowcommit.get_attribute_history(state, self.key,passive=self.passive_deletes)
                     if added or unchanged or deleted:
                         self._conditional_post_update(state, uowcommit, deleted + unchanged + added)
         else:
             for state in deplist:
-                (added, unchanged, deleted) = self.get_object_dependencies(state, uowcommit, passive=True)
+                (added, unchanged, deleted) = uowcommit.get_attribute_history(state, self.key,passive=True)
                 if added or deleted or unchanged:
                     for child in added:
                         self._synchronize(state, child, None, False, uowcommit)
@@ -299,7 +340,7 @@ class ManyToOneDP(DependencyProcessor):
         if delete:
             if self.cascade.delete:
                 for state in deplist:
-                    (added, unchanged, deleted) = self.get_object_dependencies(state, uowcommit, passive=self.passive_deletes)
+                    (added, unchanged, deleted) = uowcommit.get_attribute_history(state, self.key,passive=self.passive_deletes)
                     if deleted or unchanged:
                         for child in deleted + unchanged:
                             if child is not None and self.hasparent(child) is False:
@@ -310,7 +351,7 @@ class ManyToOneDP(DependencyProcessor):
             for state in deplist:
                 uowcommit.register_object(state)
                 if self.cascade.delete_orphan:
-                    (added, unchanged, deleted) = self.get_object_dependencies(state, uowcommit, passive=self.passive_deletes)
+                    (added, unchanged, deleted) = uowcommit.get_attribute_history(state, self.key,passive=self.passive_deletes)
                     if deleted:
                         for child in deleted:
                             if self.hasparent(child) is False:
@@ -318,6 +359,7 @@ class ManyToOneDP(DependencyProcessor):
                                 for c, m in self.mapper.cascade_iterator('delete', child):
                                     uowcommit.register_object(c._state, isdelete=True)
 
+
     def _synchronize(self, state, child, associationrow, clearkeys, uowcommit):
         source = child
         dest = state
@@ -344,7 +386,8 @@ class ManyToManyDP(DependencyProcessor):
         connection = uowcommit.transaction.connection(self.mapper)
         secondary_delete = []
         secondary_insert = []
-
+        secondary_update = []
+        
         if hasattr(self.prop, 'reverse_property'):
             reverse_dep = getattr(self.prop.reverse_property, '_dependency_processor', None)
         else:
@@ -352,7 +395,7 @@ class ManyToManyDP(DependencyProcessor):
             
         if delete:
             for state in deplist:
-                (added, unchanged, deleted) = self.get_object_dependencies(state, uowcommit, passive=self.passive_deletes)
+                (added, unchanged, deleted) = uowcommit.get_attribute_history(state, self.key,passive=self.passive_deletes)
                 if deleted or unchanged:
                     for child in deleted + unchanged:
                         if child is None or (reverse_dep and (reverse_dep, "manytomany", child, state) in uowcommit.attributes):
@@ -363,7 +406,7 @@ class ManyToManyDP(DependencyProcessor):
                         uowcommit.attributes[(self, "manytomany", state, child)] = True
         else:
             for state in deplist:
-                (added, unchanged, deleted) = self.get_object_dependencies(state, uowcommit)
+                (added, unchanged, deleted) = uowcommit.get_attribute_history(state, self.key)
                 if added or deleted:
                     for child in added:
                         if child is None or (reverse_dep and (reverse_dep, "manytomany", child, state) in uowcommit.attributes):
@@ -379,7 +422,13 @@ class ManyToManyDP(DependencyProcessor):
                         self._synchronize(state, child, associationrow, False, uowcommit)
                         uowcommit.attributes[(self, "manytomany", state, child)] = True
                         secondary_delete.append(associationrow)
-
+                        
+                if not self.passive_updates and unchanged and self._pks_changed(uowcommit, state):
+                    for child in unchanged:
+                        associationrow = {}
+                        self.syncrules.update(associationrow, state, child, "old_")
+                        secondary_update.append(associationrow)
+                    
         if secondary_delete:
             secondary_delete.sort()
             # TODO: precompile the delete/insert queries?
@@ -387,7 +436,13 @@ class ManyToManyDP(DependencyProcessor):
             result = connection.execute(statement, secondary_delete)
             if result.supports_sane_multi_rowcount() and result.rowcount != len(secondary_delete):
                 raise exceptions.ConcurrentModificationError("Deleted rowcount %d does not match number of objects deleted %d" % (result.rowcount, len(secondary_delete)))
-
+        
+        if secondary_update:
+            statement = self.secondary.update(sql.and_(*[c == sql.bindparam("old_" + c.key, type_=c.type) for c in self.secondary.c if c.key in associationrow]))
+            result = connection.execute(statement, secondary_update)
+            if result.supports_sane_multi_rowcount() and result.rowcount != len(secondary_update):
+                raise exceptions.ConcurrentModificationError("Updated rowcount %d does not match number of objects updated %d" % (result.rowcount, len(secondary_update)))
+            
         if secondary_insert:
             statement = self.secondary.insert()
             connection.execute(statement, secondary_insert)
@@ -396,7 +451,7 @@ class ManyToManyDP(DependencyProcessor):
         #print self.mapper.mapped_table.name + " " + self.key + " " + repr(len(deplist)) + " preprocess_dep isdelete " + repr(delete) + " direction " + repr(self.direction)
         if not delete:
             for state in deplist:
-                (added, unchanged, deleted) = self.get_object_dependencies(state, uowcommit, passive=True)
+                (added, unchanged, deleted) = uowcommit.get_attribute_history(state, self.key,passive=True)
                 if deleted:
                     for child in deleted:
                         if self.cascade.delete_orphan and self.hasparent(child) is False:
@@ -405,12 +460,10 @@ class ManyToManyDP(DependencyProcessor):
                                 uowcommit.register_object(c._state, isdelete=True)
 
     def _synchronize(self, state, child, associationrow, clearkeys, uowcommit):
-        dest = associationrow
-        source = None
-        if dest is None:
+        if associationrow is None:
             return
         self._verify_canload(child)
-        self.syncrules.execute(source, dest, state, child, clearkeys)
+        self.syncrules.execute(None, associationrow, state, child, clearkeys)
 
 class AssociationDP(OneToManyDP):
     def __init__(self, *args, **kwargs):
diff --git a/lib/sqlalchemy/orm/mapper.py b/lib/sqlalchemy/orm/mapper.py
index 20aa4aa64..f5e38c1c4 100644
--- a/lib/sqlalchemy/orm/mapper.py
+++ b/lib/sqlalchemy/orm/mapper.py
@@ -109,6 +109,7 @@ class Mapper(object):
         self.polymorphic_on = polymorphic_on
         self._eager_loaders = util.Set()
         self._row_translators = {}
+        self._dependency_processors = []
         
         # our 'polymorphic identity', a string name that when located in a result set row
         # indicates this Mapper should be used to construct the object instance for that row.
@@ -917,25 +918,31 @@ class Mapper(object):
             return issubclass(state.class_, self.class_)
         else:
             return state.class_ is self.class_
-    
-    def _get_state_attr_by_column(self, state, column):
+
+    def _get_col_to_prop(self, column):
         try:
-            return self._columntoproperty[column].getattr(state, column)
+            return self._columntoproperty[column]
         except KeyError:
             prop = self.__props.get(column.key, None)
             if prop:
                 raise exceptions.InvalidRequestError("Column '%s.%s' is not available, due to conflicting property '%s':%s" % (column.table.name, column.name, column.key, repr(prop)))
             else:
                 raise exceptions.InvalidRequestError("No column %s.%s is configured on mapper %s..." % (column.table.name, column.name, str(self)))
+        
+    def _get_state_attr_by_column(self, state, column):
+        return self._get_col_to_prop(column).getattr(state, column)
     
     def _set_state_attr_by_column(self, state, column, value):
-        return self._columntoproperty[column].setattr(state, value, column)
+        return self._get_col_to_prop(column).setattr(state, value, column)
         
     def _get_attr_by_column(self, obj, column):
-        return self._get_state_attr_by_column(obj._state, column)
+        return self._get_col_to_prop(column).getattr(obj._state, column)
+
+    def _get_committed_attr_by_column(self, obj, column):
+        return self._get_col_to_prop(column).getcommitted(obj._state, column)
         
     def _set_attr_by_column(self, obj, column, value):
-        self._set_state_attr_by_column(obj._state, column, value)
+        self._get_col_to_prop(column).setattr(obj._state, column, value)
 
     def save_obj(self, states, uowtransaction, postupdate=False, post_update_cols=None, single=False):
         """Issue ``INSERT`` and/or ``UPDATE`` statements for a list of objects.
@@ -991,9 +998,9 @@ class Mapper(object):
                 if self.__should_log_debug:
                     self.__log_debug("detected row switch for identity %s.  will update %s, remove %s from transaction" % (instance_key, mapperutil.state_str(state), mapperutil.instance_str(existing)))
                 uowtransaction.set_row_switch(existing)
-            if _state_has_identity(state):
-                if state.dict['_instance_key'] != instance_key:
-                    raise exceptions.FlushError("Can't change the identity of instance %s in session (existing identity: %s; new identity: %s)" % (mapperutil.state_str(state), state.dict['_instance_key'], instance_key))
+#            if _state_has_identity(state):
+#                if state.dict['_instance_key'] != instance_key:
+#                    raise exceptions.FlushError("Can't change the identity of instance %s in session (existing identity: %s; new identity: %s)" % (mapperutil.state_str(state), state.dict['_instance_key'], instance_key))
 
         inserted_objects = util.Set()
         updated_objects = util.Set()
@@ -1054,21 +1061,31 @@ class Mapper(object):
                                 (added, unchanged, deleted) = attributes.get_history(state, prop.key, passive=True)
                                 if added:
                                     hasdata = True
-                        elif col in pks:
-                            params[col._label] = mapper._get_state_attr_by_column(state, col)
                         elif mapper.polymorphic_on is not None and mapper.polymorphic_on.shares_lineage(col):
                             pass
                         else:
                             if post_update_cols is not None and col not in post_update_cols:
+                                if col in pks:
+                                    params[col._label] = mapper._get_state_attr_by_column(state, col)
                                 continue
+                                
                             prop = mapper._columntoproperty[col]
-                            (added, unchanged, deleted) = attributes.get_history(state, prop.key, passive=True)
+                            (added, unchanged, deleted) = uowtransaction.get_attribute_history(state, prop.key, passive=True, cache=False)
+                            #(added, unchanged, deleted) = attributes.get_history(state, prop.key, passive=True)
                             if added:
                                 if isinstance(added[0], sql.ClauseElement):
                                     value_params[col] = added[0]
                                 else:
                                     params[col.key] = prop.get_col_value(col, added[0])
+                                if col in pks:
+                                    if deleted:
+                                        params[col._label] = deleted[0]
+                                    else:
+                                        # row switch logic can reach us here
+                                        params[col._label] = added[0]
                                 hasdata = True
+                            elif col in pks:
+                                params[col._label] = mapper._get_state_attr_by_column(state, col)
                     if hasdata:
                         update.append((state, params, mapper, connection, value_params))
 
@@ -1233,7 +1250,7 @@ class Mapper(object):
                 if 'after_delete' in mapper.extension.methods:
                     mapper.extension.after_delete(mapper, connection, state.obj())
 
-    def register_dependencies(self, uowcommit, *args, **kwargs):
+    def register_dependencies(self, uowcommit):
         """Register ``DependencyProcessor`` instances with a
         ``unitofwork.UOWTransaction``.
 
@@ -1242,8 +1259,10 @@ class Mapper(object):
         """
 
         for prop in self.__props.values():
-            prop.register_dependencies(uowcommit, *args, **kwargs)
-
+            prop.register_dependencies(uowcommit)
+        for dep in self._dependency_processors:
+            dep.register_dependencies(uowcommit)
+            
     def cascade_iterator(self, type, state, recursive=None, halt_on=None):
         """Iterate each element and its mapper in an object graph, 
         for all relations that meet the given cascade rule.
diff --git a/lib/sqlalchemy/orm/properties.py b/lib/sqlalchemy/orm/properties.py
index 1e6d3ba7b..775bcd0a5 100644
--- a/lib/sqlalchemy/orm/properties.py
+++ b/lib/sqlalchemy/orm/properties.py
@@ -58,6 +58,9 @@ class ColumnProperty(StrategizedProperty):
     def getattr(self, state, column):
         return getattr(state.class_, self.key).impl.get(state)
 
+    def getcommitted(self, state, column):
+        return getattr(state.class_, self.key).impl.get_committed_value(state)
+
     def setattr(self, state, value, column):
         getattr(state.class_, self.key).impl.set(state, value, None)
         
@@ -99,6 +102,10 @@ class CompositeProperty(ColumnProperty):
         obj = getattr(state.class_, self.key).impl.get(state)
         return self.get_col_value(column, obj)
 
+    def getcommitted(self, state, column):
+        obj = getattr(state.class_, self.key).impl.get_committed_value(state)
+        return self.get_col_value(column, obj)
+
     def setattr(self, state, value, column):
         # TODO: test coverage for this method
         obj = getattr(state.class_, self.key).impl.get(state)
@@ -168,7 +175,7 @@ class PropertyLoader(StrategizedProperty):
     of items that correspond to a related database table.
     """
 
-    def __init__(self, argument, secondary=None, primaryjoin=None, secondaryjoin=None, entity_name=None, foreign_keys=None, foreignkey=None, uselist=None, private=False, association=None, order_by=False, attributeext=None, backref=None, is_backref=False, post_update=False, cascade=None, viewonly=False, lazy=True, collection_class=None, passive_deletes=False, remote_side=None, enable_typechecks=True, join_depth=None, strategy_class=None):
+    def __init__(self, argument, secondary=None, primaryjoin=None, secondaryjoin=None, entity_name=None, foreign_keys=None, foreignkey=None, uselist=None, private=False, association=None, order_by=False, attributeext=None, backref=None, is_backref=False, post_update=False, cascade=None, viewonly=False, lazy=True, collection_class=None, passive_deletes=False, passive_updates=True, remote_side=None, enable_typechecks=True, join_depth=None, strategy_class=None):
         self.uselist = uselist
         self.argument = argument
         self.entity_name = entity_name
@@ -185,6 +192,7 @@ class PropertyLoader(StrategizedProperty):
             util.warn_deprecated('foreignkey option is deprecated; see docs for details')
         self.collection_class = collection_class
         self.passive_deletes = passive_deletes
+        self.passive_updates = passive_updates
         self.remote_side = util.to_set(remote_side)
         self.enable_typechecks = enable_typechecks
         self._parent_join_cache = {}
@@ -214,9 +222,9 @@ class PropertyLoader(StrategizedProperty):
             # just a string was sent
             if secondary is not None:
                 # reverse primary/secondary in case of a many-to-many
-                self.backref = BackRef(backref, primaryjoin=secondaryjoin, secondaryjoin=primaryjoin)
+                self.backref = BackRef(backref, primaryjoin=secondaryjoin, secondaryjoin=primaryjoin, passive_updates=self.passive_updates)
             else:
-                self.backref = BackRef(backref, primaryjoin=primaryjoin, secondaryjoin=secondaryjoin)
+                self.backref = BackRef(backref, primaryjoin=primaryjoin, secondaryjoin=secondaryjoin, passive_updates=self.passive_updates)
         else:
             self.backref = backref
         self.is_backref = is_backref
diff --git a/lib/sqlalchemy/orm/strategies.py b/lib/sqlalchemy/orm/strategies.py
index 5a765fbd3..9adf17f42 100644
--- a/lib/sqlalchemy/orm/strategies.py
+++ b/lib/sqlalchemy/orm/strategies.py
@@ -109,7 +109,8 @@ class ColumnLoader(LoaderStrategy):
             def create_statement(instance):
                 params = {}
                 for (c, bind) in param_names:
-                    params[bind] = mapper._get_attr_by_column(instance, c)
+                    # use the "committed" (database) version to get query column values
+                    params[bind] = mapper._get_committed_attr_by_column(instance, c)
                 return (statement, params)
             
             def new_execute(instance, row, isnew, **flags):
@@ -300,7 +301,8 @@ class LazyLoader(AbstractRelationLoader):
         def visit_bindparam(bindparam):
             mapper = reverse_direction and self.parent_property.mapper or self.parent_property.parent
             if bindparam.key in bind_to_col:
-                bindparam.value = mapper._get_attr_by_column(instance, bind_to_col[bindparam.key])
+                # use the "committed" (database) version to get query column values
+                bindparam.value = mapper._get_committed_attr_by_column(instance, bind_to_col[bindparam.key])
         return visitors.traverse(criterion, clone=True, visit_bindparam=visit_bindparam)
     
     def setup_loader(self, instance, options=None, path=None):
@@ -337,7 +339,8 @@ class LazyLoader(AbstractRelationLoader):
             if self.use_get:
                 params = {}
                 for col, bind in self.lazybinds.iteritems():
-                    params[bind.key] = self.parent._get_attr_by_column(instance, col)
+                    # use the "committed" (database) version to get query column values
+                    params[bind.key] = self.parent._get_committed_attr_by_column(instance, col)
                 ident = []
                 nonnulls = False
                 for primary_key in self.select_mapper.primary_key: 
diff --git a/lib/sqlalchemy/orm/sync.py b/lib/sqlalchemy/orm/sync.py
index 2d6328514..ed263fc39 100644
--- a/lib/sqlalchemy/orm/sync.py
+++ b/lib/sqlalchemy/orm/sync.py
@@ -12,7 +12,7 @@ clause that compares column values.
 from sqlalchemy import schema, exceptions, util
 from sqlalchemy.sql import visitors, operators
 from sqlalchemy import logging
-from sqlalchemy.orm import util as mapperutil
+from sqlalchemy.orm import util as mapperutil, attributes
 
 ONETOMANY = 0
 MANYTOONE = 1
@@ -86,10 +86,21 @@ class ClauseSynchronizer(object):
     def dest_columns(self):
         return [r.dest_column for r in self.syncrules if r.dest_column is not None]
 
+    def update(self, dest, parent, child, old_prefix):
+        for rule in self.syncrules:
+            rule.update(dest, parent, child, old_prefix)
+        
     def execute(self, source, dest, obj=None, child=None, clearkeys=None):
         for rule in self.syncrules:
             rule.execute(source, dest, obj, child, clearkeys)
-
+    
+    def source_changes(self, uowcommit, source):
+        for rule in self.syncrules:
+            if rule.source_changes(uowcommit, source):
+                return True
+        else:
+            return False
+            
 class SyncRule(object):
     """An instruction indicating how to populate the objects on each
     side of a relationship.
@@ -117,8 +128,25 @@ class SyncRule(object):
         except AttributeError:
             self._dest_primary_key = self.dest_mapper is not None and self.dest_column in self.dest_mapper._pks_by_table[self.dest_column.table] and not self.dest_mapper.allow_null_pks
             return self._dest_primary_key
-
+    
+    def source_changes(self, uowcommit, source):
+        prop = self.source_mapper._columntoproperty[self.source_column]
+        (added, unchanged, deleted) = uowcommit.get_attribute_history(source, prop.key, passive=True)
+        return bool(added)
+    
+    def update(self, dest, parent, child, old_prefix):
+        if self.issecondary is False:
+            source = parent
+        elif self.issecondary is True:
+            source = child
+        oldvalue = self.source_mapper._get_committed_attr_by_column(source.obj(), self.source_column)
+        value = self.source_mapper._get_state_attr_by_column(source, self.source_column)
+        dest[self.dest_column.key] = value
+        dest[old_prefix + self.dest_column.key] = oldvalue
+        
     def execute(self, source, dest, parent, child, clearkeys):
+        # TODO: break the "dictionary" case into a separate method like 'update' above,
+        # reduce conditionals
         if source is None:
             if self.issecondary is False:
                 source = parent
diff --git a/lib/sqlalchemy/orm/unitofwork.py b/lib/sqlalchemy/orm/unitofwork.py
index e4c65a214..02c230d08 100644
--- a/lib/sqlalchemy/orm/unitofwork.py
+++ b/lib/sqlalchemy/orm/unitofwork.py
@@ -114,9 +114,18 @@ class UnitOfWork(object):
         """register the given object as 'clean' (i.e. persistent) within this unit of work, after
         a save operation has taken place."""
 
+        mapper = _state_mapper(state)
+        instance_key = mapper._identity_key_from_state(state)
+        
         if '_instance_key' not in state.dict:
-            mapper = _state_mapper(state)
-            state.dict['_instance_key'] = mapper._identity_key_from_state(state)
+            state.dict['_instance_key'] = instance_key
+            
+        elif state.dict['_instance_key'] != instance_key:
+            # primary key switch
+            self.identity_map[instance_key] = state.obj()
+            del self.identity_map[state.dict['_instance_key']]
+            state.dict['_instance_key'] = instance_key
+            
         if hasattr(state, 'insert_order'):
             delattr(state, 'insert_order')
         self.identity_map[state.dict['_instance_key']] = state.obj()
@@ -269,6 +278,33 @@ class UOWTransaction(object):
         self.attributes = {}
 
         self.logger = logging.instance_logger(self, echoflag=session.echo_uow)
+
+    def get_attribute_history(self, state, key, passive=True, cache=True):
+        hashkey = ("history", state, key)
+
+        # cache the objects, not the states; the strong reference here
+        # prevents newly loaded objects from being dereferenced during the 
+        # flush process
+        if cache and hashkey in self.attributes:
+            (added, unchanged, deleted, cached_passive) = self.attributes[hashkey]
+            # if the cached lookup was "passive" and now we want non-passive, do a non-passive
+            # lookup and re-cache
+            if cached_passive and not passive:
+                (added, unchanged, deleted) = attributes.get_history(state, key, passive=False)
+                self.attributes[hashkey] = (added, unchanged, deleted, passive)
+        else:
+            (added, unchanged, deleted) = attributes.get_history(state, key, passive=passive)
+            self.attributes[hashkey] = (added, unchanged, deleted, passive)
+
+        if added is None:
+            return (added, unchanged, deleted)
+        else:
+            return (
+                [getattr(c, '_state', c) for c in added],
+                [getattr(c, '_state', c) for c in unchanged],
+                [getattr(c, '_state', c) for c in deleted],
+                )
+
         
     def register_object(self, state, isdelete = False, listonly = False, postupdate=False, post_update_cols=None, **kwargs):
         # if object is not in the overall session, do nothing
@@ -378,7 +414,7 @@ class UOWTransaction(object):
         task = self.get_task_by_mapper(mapper)
         targettask = self.get_task_by_mapper(mapperfrom)
         up = UOWDependencyProcessor(processor, targettask)
-        task._dependencies.add(up)
+        task.dependencies.add(up)
         
     def execute(self):
         """Execute this UOWTransaction.
@@ -480,7 +516,7 @@ class UOWTask(object):
         # mapping of InstanceState -> UOWTaskElement
         self._objects = {} 
 
-        self._dependencies = util.Set()
+        self.dependencies = util.Set()
         self.cyclical_dependencies = util.Set()
 
     def polymorphic_tasks(self):
@@ -519,7 +555,7 @@ class UOWTask(object):
         used only for debugging output.
         """
 
-        return not self._objects and not self._dependencies
+        return not self._objects and not self.dependencies
             
     def append(self, state, listonly=False, isdelete=False):
         if state not in self._objects:
@@ -594,8 +630,6 @@ class UOWTask(object):
     polymorphic_todelete_objects = property(lambda self:[rec.state for rec in self.polymorphic_elements
                                           if rec.state is not None and not rec.listonly and rec.isdelete is True])
 
-    dependencies = property(lambda self:self._dependencies)
-    
     polymorphic_dependencies = _polymorphic_collection(lambda task:task.dependencies)
     
     polymorphic_cyclical_dependencies = _polymorphic_collection(lambda task:task.cyclical_dependencies)
@@ -656,7 +690,8 @@ class UOWTask(object):
                     object_to_original_task[state] = subtask
                     for dep in deps_by_targettask.get(subtask, []):
                         # is this dependency involved in one of the cycles ?
-                        if not dependency_in_cycles(dep):
+                        # (don't count the DetectKeySwitch prop)
+                        if dep.processor.no_dependencies or not dependency_in_cycles(dep):
                             continue
                         (processor, targettask) = (dep.processor, dep.targettask)
                         isdelete = taskelement.isdelete
@@ -726,7 +761,7 @@ class UOWTask(object):
         
         # stick the non-circular dependencies onto the new UOWTask
         for d in extradeplist:
-            t._dependencies.add(d)
+            t.dependencies.add(d)
         
         if head is not None:
             make_task_tree(head, t, {})
@@ -741,7 +776,7 @@ class UOWTask(object):
                 for state in t2.elements:
                     localtask.append(obj, t2.listonly, isdelete=t2._objects[state].isdelete)
                 for dep in t2.dependencies:
-                    localtask._dependencies.add(dep)
+                    localtask.dependencies.add(dep)
                 ret.insert(0, localtask)
         
         return ret
@@ -867,7 +902,7 @@ class UOWDependencyProcessor(object):
             self.processor.process_dependencies(self.targettask, [elem.state for elem in self.targettask.polymorphic_todelete_elements if elem.state is not None], trans, delete=True)
 
     def get_object_dependencies(self, state, trans, passive):
-        return self.processor.get_object_dependencies(state, trans, passive=passive)
+        return trans.get_attribute_history(state, self.processor.key, passive=passive)
 
     def whose_dependent_on_who(self, state1, state2):
         """establish which object is operationally dependent amongst a parent/child 
diff --git a/test/orm/alltests.py b/test/orm/alltests.py
index 059d7a100..3748d3f34 100644
--- a/test/orm/alltests.py
+++ b/test/orm/alltests.py
@@ -17,7 +17,8 @@ def suite():
         'orm.generative',
         'orm.lazytest1',
         'orm.assorted_eager',
-        
+
+        'orm.naturalpks',
         'orm.sessioncontext', 
         'orm.unitofwork',
         'orm.session',
diff --git a/test/orm/naturalpks.py b/test/orm/naturalpks.py
new file mode 100644
index 000000000..515c30e29
--- /dev/null
+++ b/test/orm/naturalpks.py
@@ -0,0 +1,346 @@
+import testbase
+from sqlalchemy import *
+from sqlalchemy.orm import *
+from sqlalchemy import exceptions
+
+from testlib.fixtures import *
+from testlib import *
+
+"""test primary key changing capabilities and passive/non-passive cascading updates."""
+
+class NaturalPKTest(ORMTest):
+    def define_tables(self, metadata):
+        global users, addresses, items, users_to_items
+        
+        users = Table('users', metadata,
+            Column('username', String(50), primary_key=True),
+            Column('fullname', String(100)))
+            
+        addresses = Table('addresses', metadata,
+            Column('email', String(50), primary_key=True),
+            Column('username', String(50), ForeignKey('users.username', onupdate="cascade")))
+            
+        items = Table('items', metadata,
+            Column('itemname', String(50), primary_key=True),
+            Column('description', String(100)))
+            
+        users_to_items = Table('userstoitems', metadata,
+            Column('username', String(50), ForeignKey('users.username', onupdate='cascade'), primary_key=True),
+            Column('itemname', String(50), ForeignKey('items.itemname', onupdate='cascade'), primary_key=True),
+        )
+        
+    def test_entity(self):
+        mapper(User, users)
+        
+        sess = create_session()
+        u1 = User(username='jack', fullname='jack')
+        
+        sess.save(u1)
+        sess.flush()
+        assert sess.get(User, 'jack') is u1
+        
+        u1.username = 'ed'
+        sess.flush()
+        
+        def go():
+            assert sess.get(User, 'ed') is u1
+        self.assert_sql_count(testbase.db, go, 0)
+
+        assert sess.get(User, 'jack') is None
+        
+        sess.clear()
+        u1 = sess.query(User).get('ed')
+        self.assertEquals(User(username='ed', fullname='jack'), u1)
+    
+    @testing.unsupported('sqlite','mysql')
+    def test_onetomany_passive(self):
+        self._test_onetomany(True)
+    
+    def test_onetomany_nonpassive(self):
+        self._test_onetomany(False)
+        
+    def _test_onetomany(self, passive_updates):
+        mapper(User, users, properties={
+            'addresses':relation(Address, passive_updates=passive_updates)
+        })
+        mapper(Address, addresses)
+    
+        sess = create_session()
+        u1 = User(username='jack', fullname='jack')
+        u1.addresses.append(Address(email='jack1'))
+        u1.addresses.append(Address(email='jack2'))
+        sess.save(u1)
+        sess.flush()
+    
+        assert sess.get(Address, 'jack1') is u1.addresses[0]
+    
+        u1.username = 'ed'
+        sess.flush()
+        assert u1.addresses[0].username == 'ed'
+        
+        sess.clear()
+        self.assertEquals([Address(username='ed'), Address(username='ed')], sess.query(Address).all())
+        
+        u1 = sess.get(User, 'ed')
+        u1.username = 'jack'
+        def go():
+            sess.flush()
+        if not passive_updates:
+            self.assert_sql_count(testbase.db, go, 4) # test passive_updates=False; load addresses, update user, update 2 addresses
+        else:
+            self.assert_sql_count(testbase.db, go, 1) # test passive_updates=True; update user
+        sess.clear()
+        assert User(username='jack', addresses=[Address(username='jack'), Address(username='jack')]) == sess.get(User, 'jack')
+    
+        u1 = sess.get(User, 'jack')
+        u1.addresses = []
+        u1.username = 'fred'
+        sess.flush()
+        sess.clear()
+        assert sess.get(Address, 'jack1').username is None
+        u1 = sess.get(User, 'fred')
+        self.assertEquals(User(username='fred', fullname='jack'), u1)
+
+    @testing.unsupported('sqlite', 'mysql')
+    def test_manytoone_passive(self):
+        self._test_manytoone(True)
+
+    def test_manytoone_nonpassive(self):
+        self._test_manytoone(False)
+    
+    def _test_manytoone(self, passive_updates):
+        mapper(User, users)
+        mapper(Address, addresses, properties={
+            'user':relation(User, passive_updates=passive_updates)
+        })
+        
+        sess = create_session()
+        a1 = Address(email='jack1')
+        a2 = Address(email='jack2')
+        
+        u1 = User(username='jack', fullname='jack')
+        a1.user = u1
+        a2.user = u1
+        sess.save(a1)
+        sess.save(a2)
+        sess.flush()
+        
+        u1.username = 'ed'
+        
+        print id(a1), id(a2), id(u1)
+        print u1._state.parents
+        def go():
+            sess.flush()
+        if passive_updates:
+            self.assert_sql_count(testbase.db, go, 1)
+        else:
+            self.assert_sql_count(testbase.db, go, 3)
+        
+        def go():
+            sess.flush()
+        self.assert_sql_count(testbase.db, go, 0)
+        
+        assert a1.username == a2.username == 'ed'
+        sess.clear()
+        self.assertEquals([Address(username='ed'), Address(username='ed')], sess.query(Address).all())
+
+    @testing.unsupported('sqlite', 'mysql')
+    def test_bidirectional_passive(self):
+        self._test_bidirectional(True)
+
+    def test_bidirectional_nonpassive(self):
+        self._test_bidirectional(False)
+
+    def _test_bidirectional(self, passive_updates):
+        mapper(User, users)
+        mapper(Address, addresses, properties={
+            'user':relation(User, passive_updates=passive_updates, backref='addresses')
+        })
+
+        sess = create_session()
+        a1 = Address(email='jack1')
+        a2 = Address(email='jack2')
+
+        u1 = User(username='jack', fullname='jack')
+        a1.user = u1
+        a2.user = u1
+        sess.save(a1)
+        sess.save(a2)
+        sess.flush()
+
+        u1.username = 'ed'
+        (ad1, ad2) = sess.query(Address).all()
+        self.assertEquals([Address(username='jack'), Address(username='jack')], [ad1, ad2])
+        def go():
+            sess.flush()
+        if passive_updates:
+            self.assert_sql_count(testbase.db, go, 1)
+        else:
+            self.assert_sql_count(testbase.db, go, 3)
+        self.assertEquals([Address(username='ed'), Address(username='ed')], [ad1, ad2])
+        sess.clear()
+        self.assertEquals([Address(username='ed'), Address(username='ed')], sess.query(Address).all())
+
+        u1 = sess.get(User, 'ed')
+        assert len(u1.addresses) == 2    # load addresses
+        u1.username = 'fred'
+        print "--------------------------------"
+        def go():
+            sess.flush()
+        # check that the passive_updates is on on the other side
+        if passive_updates:
+            self.assert_sql_count(testbase.db, go, 1)
+        else:
+            self.assert_sql_count(testbase.db, go, 3)
+        sess.clear()
+        self.assertEquals([Address(username='fred'), Address(username='fred')], sess.query(Address).all())
+        
+        
+    @testing.unsupported('sqlite', 'mysql')
+    def test_manytomany_passive(self):
+        self._test_manytomany(True)
+    
+    def test_manytomany_nonpassive(self):
+        self._test_manytomany(False)
+        
+    def _test_manytomany(self, passive_updates):
+        mapper(User, users, properties={
+            'items':relation(Item, secondary=users_to_items, backref='users', passive_updates=passive_updates)
+        })
+        mapper(Item, items)
+        
+        sess = create_session()
+        u1 = User(username='jack')
+        u2 = User(username='fred')
+        i1 = Item(itemname='item1')
+        i2 = Item(itemname='item2')
+        
+        u1.items.append(i1)
+        u1.items.append(i2)
+        i2.users.append(u2)
+        sess.save(u1)
+        sess.save(u2)
+        sess.flush()
+
+        r = sess.query(Item).all()
+        # fixtures.Base can't handle a comparison with the backrefs involved....
+        self.assertEquals(Item(itemname='item1'), r[0])
+        self.assertEquals(['jack'], [u.username for u in r[0].users])
+        self.assertEquals(Item(itemname='item2'), r[1])
+        self.assertEquals(['jack', 'fred'], [u.username for u in r[1].users])
+        
+        u2.username='ed'
+        def go():
+            sess.flush()
+        go()
+        def go():
+            sess.flush()
+        self.assert_sql_count(testbase.db, go, 0)
+        
+        sess.clear()
+        r = sess.query(Item).all()
+        self.assertEquals(Item(itemname='item1'), r[0])
+        self.assertEquals(['jack'], [u.username for u in r[0].users])
+        self.assertEquals(Item(itemname='item2'), r[1])
+        self.assertEquals(['ed', 'jack'], sorted([u.username for u in r[1].users]))
+        
+class SelfRefTest(ORMTest):
+    def define_tables(self, metadata):
+        global nodes, Node
+        
+        nodes = Table('nodes', metadata,
+            Column('name', String(50), primary_key=True),
+            Column('parent', String(50), ForeignKey('nodes.name', onupdate='cascade'))
+            )
+            
+        class Node(Base):
+            pass
+    
+    def test_onetomany(self):
+        mapper(Node, nodes, properties={
+            'children':relation(Node, backref=backref('parentnode', remote_side=nodes.c.name, passive_updates=False), passive_updates=False)
+        })
+        
+        sess = create_session()
+        n1 = Node(name='n1')
+        n1.children.append(Node(name='n11'))
+        n1.children.append(Node(name='n12'))
+        n1.children.append(Node(name='n13'))
+        sess.save(n1)
+        sess.flush()
+        
+        n1.name = 'new n1'
+        sess.flush()
+        self.assertEquals(n1.children[1].parent, 'new n1')
+        self.assertEquals(['new n1', 'new n1', 'new n1'], [n.parent for n in sess.query(Node).filter(Node.name.in_(['n11', 'n12', 'n13']))])
+        
+        
+class NonPKCascadeTest(ORMTest):
+    def define_tables(self, metadata):
+        global users, addresses
+
+        users = Table('users', metadata,
+            Column('id', Integer, primary_key=True),
+            Column('username', String(50), unique=True),
+            Column('fullname', String(100)))
+
+        addresses = Table('addresses', metadata,
+            Column('id', Integer, primary_key=True),
+            Column('email', String(50)),
+            Column('username', String(50), ForeignKey('users.username', onupdate="cascade")))
+
+    @testing.unsupported('sqlite','mysql')
+    def test_onetomany_passive(self):
+        self._test_onetomany(True)
+
+    def test_onetomany_nonpassive(self):
+        self._test_onetomany(False)
+
+    def _test_onetomany(self, passive_updates):
+        mapper(User, users, properties={
+            'addresses':relation(Address, passive_updates=passive_updates)
+        })
+        mapper(Address, addresses)
+
+        sess = create_session()
+        u1 = User(username='jack', fullname='jack')
+        u1.addresses.append(Address(email='jack1'))
+        u1.addresses.append(Address(email='jack2'))
+        sess.save(u1)
+        sess.flush()
+        a1 = u1.addresses[0]
+        
+        assert sess.get(Address, a1.id) is u1.addresses[0]
+
+        u1.username = 'ed'
+        sess.flush()
+        assert u1.addresses[0].username == 'ed'
+
+        sess.clear()
+        self.assertEquals([Address(username='ed'), Address(username='ed')], sess.query(Address).all())
+
+        u1 = sess.get(User, u1.id)
+        u1.username = 'jack'
+        def go():
+            sess.flush()
+        if not passive_updates:
+            self.assert_sql_count(testbase.db, go, 4) # test passive_updates=False; load addresses, update user, update 2 addresses
+        else:
+            self.assert_sql_count(testbase.db, go, 1) # test passive_updates=True; update user
+        sess.clear()
+        assert User(username='jack', addresses=[Address(username='jack'), Address(username='jack')]) == sess.get(User, u1.id)
+
+        u1 = sess.get(User, u1.id)
+        u1.addresses = []
+        u1.username = 'fred'
+        sess.flush()
+        sess.clear()
+        assert sess.get(Address, a1.id).username is None
+        u1 = sess.get(User, u1.id)
+        self.assertEquals(User(username='fred', fullname='jack'), u1)
+
+        
+if __name__ == '__main__':
+    testbase.main()
+        
+        
diff --git a/test/orm/unitofwork.py b/test/orm/unitofwork.py
index 2fa286ba8..0464e2364 100644
--- a/test/orm/unitofwork.py
+++ b/test/orm/unitofwork.py
@@ -458,24 +458,6 @@ class PKTest(ORMTest):
         e.data = 'some more data'
         Session.commit()
 
-    @engines.assert_conns_closed
-    def test_pksimmutable(self):
-        class Entry(object):
-            pass
-        mapper(Entry, table)
-        e = Entry()
-        e.multi_id=5
-        e.multi_rev=5
-        e.name='somename'
-        Session.commit()
-        e.multi_rev=6
-        e.name = 'someothername'
-        try:
-            Session.commit()
-            assert False
-        except exceptions.FlushError, fe:
-            assert str(fe) == "Can't change the identity of instance Entry@%s in session (existing identity: (%s, (5, 5), None); new identity: (%s, (5, 6), None))" % (hex(id(e)), repr(e.__class__), repr(e.__class__)), str(fe)
-            
 class ForeignPKTest(ORMTest):
     """tests mapper detection of the relationship direction when parent/child tables are joined on their
     primary keys"""