Merge lp:~edoardo-serra/storm/select-for-update-support into lp:storm

[Edoardo Serra]

This branches implements the SELECT ... FOR UPDATE

[Jamu Kakar (jkakar)]

Edoardo, thanks for this branch! Before we spend time reviewing the
changes, have you signed the Canonical copyright assignment forms?
This process will need to be completed before we can accept changes.

[Edoardo Serra (edoardo-serra)]

Hi Jamu, sorry I didn't find any mention of this on the Wiki.

I just accepdet the form via email.

Regards

On Sep 10, 2010, at 4:52 PM, Jamu Kakar wrote:

> Edoardo, thanks for this branch! Before we spend time reviewing the
> changes, have you signed the Canonical copyright assignment forms?
> This process will need to be completed before we can accept changes.
>
> --
> https://code.launchpad.net/~edoardo-serra/storm/select-for-update-support/+merge/35112
> You are the owner of lp:~edoardo-serra/storm/select-for-update-support.

[James Henstridge (jamesh)]

This isn't a full review, but here are a few initial comments:

1. If "SELECT FOR UPDATE" is incompatible with "GROUP BY", perhaps it would make sense to check for that in Select's compile function and raise ExprError there rather than trying to protect all the ResultSet entry points that could cause this problem.

2. You've added a comment asking whether it makes sense to remove the "FOR UPDATE" clause for ResultSet.is_empty(). If is_empty() returns True, then no rows would be locked if "FOR UPDATE" was left in. If it returns False, the program is likely to request those rows if it actually wants them locked. So I don't think the comment is necessary.

3. This doesn't need to go in your branch, but I wonder if it would be worth storing the face that a row has been locked in its obj_info, so that Store.get(..., for_update=True) can avoid a query if the row has already been locked? This could be set in Store._load_object() and unset by Store.invalidate().

[James Henstridge (jamesh)]

One other point: it is probably worth using an API that can support "FOR SHARE" read locks in addition to "FOR UPDATE" write locks.

Unmerged revisions

380. By Edoardo Serra <eserra@barbera> on 2010-09-09

SELECT ... FOR UPDATE implemented in Store.get

379. By Edoardo Serra <eserra@barbera> on 2010-09-08

Unit tests

378. By Edoardo Serra <eserra@barbera> on 2010-09-08

SELECT ... FOR UPDATE is not compatible with GROUP BY

377. By Edoardo Serra <eserra@barbera> on 2010-09-08

SQLite does not support SELECT ... FOR UPDATE

376. By Edoardo Serra <eserra@barbera> on 2010-09-08

Starting implementation of SELECT ... FOR UPATE

=== modified file 'storm/databases/sqlite.py'
--- storm/databases/sqlite.py	2010-06-14 11:52:25 +0000
+++ storm/databases/sqlite.py	2010-09-10 14:44:22 +0000
@@ -49,6 +49,10 @@
 def compile_select_sqlite(compile, select, state):
     if select.offset is not Undef and select.limit is Undef:
         select.limit = sys.maxint
+    # SQLite does not support SELECT ... FOR UPDATE
+    # Can we just ignore it?
+    if select.for_update:
+        select.for_update = False
     statement = compile_select(compile, select, state)
     if state.context is SELECT:
         # SQLite breaks with (SELECT ...) UNION (SELECT ...), so we

=== modified file 'storm/expr.py'
--- storm/expr.py	2009-11-02 11:11:20 +0000
+++ storm/expr.py	2010-09-10 14:44:22 +0000
@@ -636,12 +636,13 @@
 
 class Select(Expr):
     __slots__ = ("columns", "where", "tables", "default_tables", "order_by",
-                 "group_by", "limit", "offset", "distinct", "having")
+                 "group_by", "limit", "offset", "distinct", "having",
+                 'for_update')
 
     def __init__(self, columns, where=Undef,
                  tables=Undef, default_tables=Undef,
-                 order_by=Undef, group_by=Undef,
-                 limit=Undef, offset=Undef, distinct=False, having=Undef):
+                 order_by=Undef, group_by=Undef, limit=Undef, 
+                 offset=Undef, distinct=False, having=Undef, for_update=False):
         self.columns = columns
         self.where = where
         self.tables = tables
@@ -652,6 +653,7 @@
         self.offset = offset
         self.distinct = distinct
         self.having = having
+        self.for_update = for_update
 
 @compile.when(Select)
 def compile_select(compile, select, state):
@@ -680,6 +682,8 @@
         tokens.append(" LIMIT %d" % select.limit)
     if select.offset is not Undef:
         tokens.append(" OFFSET %d" % select.offset)
+    if select.for_update:
+        tokens.append(" FOR UPDATE")
     if has_tables(state, select):
         state.context = TABLE
         state.push("parameters", [])

=== modified file 'storm/store.py'
--- storm/store.py	2010-09-01 15:00:42 +0000
+++ storm/store.py	2010-09-10 14:44:22 +0000
@@ -137,13 +137,16 @@
         self.invalidate()
         self._connection.rollback()
 
-    def get(self, cls, key):
+    def get(self, cls, key, for_update=False):
         """Get object of type cls with the given primary key from the database.
 
         If the object is alive the database won't be touched.
 
+        If the FOR UPDATE clause is used, the query is always executed
+
         @param cls: Class of the object to be retrieved.
         @param key: Primary key of object. May be a tuple for composed keys.
+        @param for_update: Indicates wether using the clause FOR UPDATE
 
         @return: The object found with the given primary key, or None
             if no object is found.
@@ -165,20 +168,22 @@
                 variable = column.variable_factory(value=variable)
             primary_vars.append(variable)
 
-        primary_values = tuple(var.get(to_db=True) for var in primary_vars)
-        obj_info = self._alive.get((cls_info.cls, primary_values))
-        if obj_info is not None:
-            if obj_info.get("invalidated"):
-                try:
-                    self._validate_alive(obj_info)
-                except LostObjectError:
-                    return None
-            return self._get_object(obj_info)
+        if not for_update:
+            primary_values = tuple(var.get(to_db=True) for var in primary_vars)
+            obj_info = self._alive.get((cls_info.cls, primary_values))
+            if obj_info is not None:
+                if obj_info.get("invalidated"):
+                    try:
+                        self._validate_alive(obj_info)
+                    except LostObjectError:
+                        return None
+                return self._get_object(obj_info)
 
         where = compare_columns(cls_info.primary_key, primary_vars)
 
         select = Select(cls_info.columns, where,
-                        default_tables=cls_info.table, limit=1)
+                        default_tables=cls_info.table, limit=1,
+                        for_update=for_update)
 
         result = self._connection.execute(select)
         values = result.get_one()
@@ -921,6 +926,7 @@
         self._distinct = False
         self._group_by = Undef
         self._having = Undef
+        self._for_update = False
 
     def copy(self):
         """Return a copy of this ResultSet object, with the same configuration.
@@ -933,7 +939,7 @@
             result_set._select = copy(self._select)
         return result_set
 
-    def config(self, distinct=None, offset=None, limit=None):
+    def config(self, distinct=None, offset=None, limit=None, for_update=None):
         """Configure this result object in-place. All parameters are optional.
 
         @param distinct: Boolean enabling/disabling usage of the DISTINCT
@@ -952,6 +958,8 @@
             self._offset = offset
         if limit is not None:
             self._limit = limit
+        if for_update is not None:
+            self._for_update = for_update
         return self
 
     def _get_select(self):
@@ -967,7 +975,7 @@
         return Select(columns, self._where, self._tables, default_tables,
                       self._order_by, offset=self._offset, limit=self._limit,
                       distinct=self._distinct, group_by=self._group_by,
-                      having=self._having)
+                      having=self._having, for_update=self._for_update)
 
     def _load_objects(self, result, values):
         return self._find_spec.load_objects(self._store, result, values)
@@ -1054,6 +1062,9 @@
         subselect = self._get_select()
         subselect.limit = 1
         subselect.order_by = Undef
+        # Should we really strip FOR UPDATE clause or should we raise a
+        # FeatureError?
+        subselect.for_update = False
         select = Select(1, tables=Alias(subselect, "_tmp"), limit=1)
         result = self._store._connection.execute(select)
         return (not result.get_one())
@@ -1162,6 +1173,15 @@
         self._order_by = args or Undef
         return self
 
+    def for_update(self):
+        """Acquire an exclusive row-level lock on the rows in the ResultSet
+        """
+        if self._group_by is not Undef:
+            raise FeatureError("SELECT ... FOR UPDATE isn't supported with a "
+                               " GROUP BY clause ")
+        self._for_update = True
+        return self
+
     def remove(self):
         """Remove all rows represented by this ResultSet from the database.
 
@@ -1193,6 +1213,9 @@
         if self._select is not Undef:
             raise FeatureError("Grouping isn't supported with "
                                "set expressions (unions, etc)")
+        if self._for_update:
+            raise FeatureError("Grouping isn't supported with "
+                               "FOR UPDATE clause")
 
         find_spec = FindSpec(expr)
         columns, dummy = find_spec.get_columns_and_tables()
@@ -1216,6 +1239,9 @@
         if self._group_by is not Undef:
             raise FeatureError("Single aggregates aren't supported after a "
                                " GROUP BY clause ")
+        if self._for_update:
+            raise FeatureError("Single aggregates aren't supported with"
+                               " FOR UPDATE clause ")
         columns, default_tables = self._find_spec.get_columns_and_tables()
         if (self._select is Undef and not self._distinct and
             self._offset is Undef and self._limit is Undef):
@@ -1471,6 +1497,8 @@
         """
         if isinstance(other, EmptyResultSet):
             return self
+        if self._for_update or other._for_update:
+            raise FeatureError("SELECT FOR UPDATE is not allowed with UNION")
         return self._set_expr(Union, other, all)
 
     def difference(self, other, all=False):
@@ -1480,6 +1508,8 @@
         """
         if isinstance(other, EmptyResultSet):
             return self
+        if self._for_update or other._for_update:
+            raise FeatureError("SELECT FOR UPDATE is not allowed with EXCEPT")
         return self._set_expr(Except, other, all)
 
     def intersection(self, other, all=False):
@@ -1489,6 +1519,9 @@
         """
         if isinstance(other, EmptyResultSet):
             return other
+        if self._for_update or other._for_update:
+            raise FeatureError("SELECT FOR UPDATE is not allowed with "
+                               "INTERSECT")
         return self._set_expr(Intersect, other, all)
 
 
@@ -1516,7 +1549,7 @@
         result = EmptyResultSet(self._order_by)
         return result
 
-    def config(self, distinct=None, offset=None, limit=None):
+    def config(self, distinct=None, offset=None, limit=None, for_update=None):
         pass
 
     def __iter__(self):
@@ -1552,6 +1585,10 @@
         self._order_by = True
         return self
 
+    def for_update(self):
+        self._for_update = True
+        return self
+
     def remove(self):
         return 0
 

=== modified file 'storm/zope/interfaces.py'
--- storm/zope/interfaces.py	2008-11-27 09:37:56 +0000
+++ storm/zope/interfaces.py	2010-09-10 14:44:22 +0000
@@ -93,6 +93,9 @@
     def order_by(*args):
         """Order the result set based on expressions in C{args}."""
 
+    def for_update():
+        """Acquire an exclusive row-level lock on the rows in the ResultSet"""
+
     def count(column=Undef, distinct=False):
         """Returns the number of rows in the result set.
 

=== modified file 'tests/expr.py'
--- tests/expr.py	2009-11-02 11:11:20 +0000
+++ tests/expr.py	2010-09-10 14:44:22 +0000
@@ -65,9 +65,11 @@
         self.assertEquals(expr.limit, Undef)
         self.assertEquals(expr.offset, Undef)
         self.assertEquals(expr.distinct, False)
+        self.assertEquals(expr.having, Undef)
+        self.assertEquals(expr.for_update, False)
 
     def test_select_constructor(self):
-        objects = [object() for i in range(9)]
+        objects = [object() for i in range(11)]
         expr = Select(*objects)
         self.assertEquals(expr.columns, objects[0])
         self.assertEquals(expr.where, objects[1])
@@ -78,6 +80,8 @@
         self.assertEquals(expr.limit, objects[6])
         self.assertEquals(expr.offset, objects[7])
         self.assertEquals(expr.distinct, objects[8])
+        self.assertEquals(expr.having, objects[9])
+        self.assertEquals(expr.for_update, objects[10])
 
     def test_insert_default(self):
         expr = Insert(None)
@@ -657,6 +661,14 @@
                           'SELECT DISTINCT column1, column2 FROM "table 1"')
         self.assertEquals(state.parameters, [])
 
+    def test_select_for_update(self):
+        expr = Select([column1, column2], Undef, [table1], for_update=True)
+        state = State()
+        statement = compile(expr, state)
+        self.assertEquals(statement,
+                          'SELECT column1, column2 FROM "table 1" FOR UPDATE')
+        self.assertEquals(state.parameters, [])
+
     def test_select_where(self):
         expr = Select([column1, Func1()],
                       Func1(),

=== modified file 'tests/store/base.py'
--- tests/store/base.py	2010-09-01 09:02:41 +0000
+++ tests/store/base.py	2010-09-10 14:44:22 +0000
@@ -321,6 +321,18 @@
         foo = self.store.get(Foo, 40)
         self.assertEquals(foo, None)
 
+    def test_get_for_update(self):
+        stream = StringIO()
+        self.addCleanup(debug, False)
+        debug(True, stream)
+
+        foo = self.store.get(Foo, 10, for_update=True)
+        self.assertEquals(foo.id, 10)
+        self.assertEquals(foo.title, "Title 30")
+        if self.__class__.__name__.startswith("SQLite"):
+            return
+        self.assertIn("FOR UPDATE", stream.getvalue())
+
     def test_get_cached(self):
         foo = self.store.get(Foo, 10)
         self.assertTrue(self.store.get(Foo, 10) is foo)
@@ -332,6 +344,13 @@
         self.store.get(Foo, 10)
         self.store._connection = connection
 
+    def test_wb_get_cached_for_update_needa_connection(self):
+        foo = self.store.get(Foo, 10)
+        connection = self.store._connection
+        self.store._connection = None
+        self.assertRaises(AttributeError, self.store.get, Foo, 10, for_update=True)
+        self.store._connection = connection
+
     def test_cache_cleanup(self):
         # Disable the cache, which holds strong references.
         self.get_cache(self.store).set_size(0)
@@ -569,6 +588,21 @@
         self.assertEqual(True, result.is_empty())
         self.assertNotIn("ORDER BY", stream.getvalue())
 
+    def test_is_empty_strips_for_update(self):
+        """
+        L{ResultSet.is_empty} strips the C{FOR UPDATE} clause, if one is
+        present, since it's not supported in subqueries and thre is no reason
+        to acquire a row-level lock while testing if a ResultSet is empty
+        """
+        stream = StringIO()
+        self.addCleanup(debug, False)
+        debug(True, stream)
+
+        result = self.store.find(Foo, Foo.id == 300).for_update()
+        result.order_by(Foo.id)
+        self.assertEqual(True, result.is_empty())
+        self.assertNotIn("FOR UPDATE", stream.getvalue())
+
     def test_is_empty_with_composed_key(self):
         result = self.store.find(Link, foo_id=300, bar_id=3000)
         self.assertEquals(result.is_empty(), True)
@@ -592,6 +626,23 @@
                           (30, "Title 10"),
                          ])
 
+    def test_find_for_update(self):
+        stream = StringIO()
+        self.addCleanup(debug, False)
+        debug(True, stream)
+
+        result = self.store.find(Foo).for_update()
+        lst = [(foo.id, foo.title) for foo in result]
+        lst.sort()
+        self.assertEquals(lst, [
+                          (10, "Title 30"),
+                          (20, "Title 20"),
+                          (30, "Title 10"),
+                         ])
+        if self.__class__.__name__.startswith("SQLite"):
+            return
+        self.assertIn("FOR UPDATE", stream.getvalue())
+
     def test_find_from_cache(self):
         foo = self.store.get(Foo, 10)
         self.assertTrue(self.store.find(Foo, id=10).one() is foo)
@@ -940,6 +991,11 @@
     def test_find_count(self):
         self.assertEquals(self.store.find(Foo).count(), 3)
 
+    def test_find_count_for_update(self):
+        result = self.store.find(Foo)
+        result.for_update()
+        self.assertRaises(FeatureError, result.count, 3)
+
     def test_find_count_after_slice(self):
         """
         When we slice a ResultSet obtained after a set operation (like union),
@@ -1007,6 +1063,11 @@
     def test_find_max(self):
         self.assertEquals(self.store.find(Foo).max(Foo.id), 30)
 
+    def test_find_max_for_update(self):
+        result = self.store.find(Foo)
+        result.for_update()
+        self.assertRaises(FeatureError, result.max, Foo.id)
+
     def test_find_max_expr(self):
         self.assertEquals(self.store.find(Foo).max(Foo.id + 1), 31)
 
@@ -1028,6 +1089,11 @@
     def test_find_min(self):
         self.assertEquals(self.store.find(Foo).min(Foo.id), 10)
 
+    def test_find_min_for_update(self):
+        result = self.store.find(Foo)
+        result.for_update()
+        self.assertRaises(FeatureError, result.min, Foo.id)
+
     def test_find_min_expr(self):
         self.assertEquals(self.store.find(Foo).min(Foo.id - 1), 9)
 
@@ -1049,6 +1115,11 @@
     def test_find_avg(self):
         self.assertEquals(self.store.find(Foo).avg(Foo.id), 20)
 
+    def test_find_avg_for_update(self):
+        result = self.store.find(Foo)
+        result.for_update()
+        self.assertRaises(FeatureError, result.avg, Foo.id)
+
     def test_find_avg_expr(self):
         self.assertEquals(self.store.find(Foo).avg(Foo.id + 10), 30)
 
@@ -1062,6 +1133,11 @@
     def test_find_sum(self):
         self.assertEquals(self.store.find(Foo).sum(Foo.id), 60)
 
+    def test_find_sum_for_update(self):
+        result = self.store.find(Foo)
+        result.for_update()
+        self.assertRaises(FeatureError, result.sum, Foo.id)
+
     def test_find_sum_expr(self):
         self.assertEquals(self.store.find(Foo).sum(Foo.id * 2), 120)
 
@@ -1546,6 +1622,11 @@
         result = list(result)
         self.assertEquals(result, [(2L, 2L), (2L, 2L), (2L, 3L), (3L, 6L)])
 
+    def test_find_group_by_for_update(self):
+        result = self.store.find((Count(FooValue.id), Sum(FooValue.value1)))
+        result.group_by(FooValue.value2)
+        self.assertRaises(FeatureError, result.for_update)
+
     def test_find_group_by_table(self):
         result = self.store.find(
             (Sum(FooValue.value2), Foo), Foo.id == FooValue.foo_id)
@@ -2271,6 +2352,13 @@
         self.assertEquals(foo.id, 20)
         self.assertEquals(foo.title, "Title 20")
 
+    def test_sub_select_for_update(self):
+        foo = self.store.find(Foo, Foo.id == Select(SQL("20"))
+                             ).for_update().one()
+        self.assertTrue(foo)
+        self.assertEquals(foo.id, 20)
+        self.assertEquals(foo.title, "Title 20")
+
     def test_cache_has_improper_object(self):
         foo = self.store.get(Foo, 20)
         self.store.remove(foo)
@@ -5472,6 +5560,12 @@
                           (30, "Title 10"),
                          ])
 
+    def test_result_union_for_update(self):
+        result1 = self.store.find(Foo, id=30).for_update()
+        result2 = self.store.find(Foo, id=10)
+        self.assertRaises(FeatureError, result1.union, result2)
+        self.assertRaises(FeatureError, result2.union, result1)
+
     def test_result_union_duplicated(self):
         result1 = self.store.find(Foo, id=30)
         result2 = self.store.find(Foo, id=30)
@@ -5544,6 +5638,12 @@
                           (30, "Title 10"),
                          ])
 
+    def test_result_difference_for_update(self):
+        result1 = self.store.find(Foo).for_update()
+        result2 = self.store.find(Foo, id=10)
+        self.assertRaises(FeatureError, result1.difference, result2)
+        self.assertRaises(FeatureError, result2.difference, result1)
+
     def test_result_difference_with_empty(self):
         if self.__class__.__name__.startswith("MySQL"):
             return
@@ -5605,6 +5705,12 @@
                           (30, "Title 10"),
                          ])
 
+    def test_result_intersection_for_update(self):
+        result1 = self.store.find(Foo).for_update()
+        result2 = self.store.find(Foo, Foo.id.is_in((10, 30)))
+        self.assertRaises(FeatureError, result1.intersection, result2)
+        self.assertRaises(FeatureError, result2.intersection, result1)
+
     def test_result_intersection_with_empty(self):
         if self.__class__.__name__.startswith("MySQL"):
             return
@@ -6048,6 +6154,10 @@
         self.assertEquals(self.result.order_by(Foo.title), self.result)
         self.assertEquals(self.empty.order_by(Foo.title), self.empty)
 
+    def test_for_update(self):
+        self.assertEquals(self.result.for_update(), self.result)
+        self.assertEquals(self.empty.for_update(), self.empty)
+
     def test_remove(self):
         self.assertEquals(self.result.remove(), 0)
         self.assertEquals(self.empty.remove(), 0)