Merge pull request pandas-dev#509 from manahl/issue-508

Issue pandas-dev#508 - List symbols returning deleted symbols

Author: aflag

CHANGES.md

@@ -2,6 +2,7 @@
 
 ### 1.61
   * Feature: #288 Mapping reads and writes over chunks in chunkstore
+  * Bugfix: #508 VersionStore: list_symbols and read now always returns latest version
 
 ### 1.60 (2018-2-13)
   * Bugfix: #503 ChunkStore: speedup check for -1 segments

arctic/store/version_store.py

@@ -149,7 +149,7 @@ def list_symbols(self, all_symbols=False, snapshot=None, regex=None, **kwargs):
             pipeline.append({'$match': query})
         pipeline.extend([
                         # Id is by insert time which matches version order
-                        {'$sort': {'_id':-1}},
+                        {'$sort': bson.SON([('symbol', pymongo.DESCENDING), ('version', pymongo.DESCENDING)])},
                         # Group by 'symbol'
                         {'$group': {'_id': '$symbol',
                                     'deleted': {'$first': '$metadata.deleted'},
@@ -422,7 +422,7 @@ def _read_metadata(self, symbol, as_of=None, read_preference=None):
                 as_of = as_of.replace(tzinfo=mktz())
             _version = versions_coll.find_one({'symbol': symbol,
                                                '_id': {'$lt': bson.ObjectId.from_datetime(as_of + timedelta(seconds=1))}},
-                                              sort=[('_id', pymongo.DESCENDING)])
+                                              sort=[('symbol', pymongo.DESCENDING), ('version', pymongo.DESCENDING)])
         else:
             # Backward compatibility - as of is a version number
             _version = versions_coll.find_one({'symbol': symbol, 'version': as_of})

tests/integration/store/test_version_store.py

@@ -1,5 +1,6 @@
 import bson
 import six
+import struct
 from bson.son import SON
 from datetime import datetime as dt, timedelta as dtd
 import pandas as pd
@@ -8,7 +9,7 @@
 from pymongo.read_preferences import ReadPreference
 from pymongo.server_type import SERVER_TYPE
 from datetime import datetime
-from mock import patch
+from mock import Mock, patch
 import time
 import pytest
 import numpy as np
@@ -151,6 +152,28 @@ def test_read_metadata(library):
     assert after.data is None
 
 
+def test_read_metadata_newer_version_with_lower_id(library):
+    now_timestamp = int(time.time())
+    now = struct.pack(">i", now_timestamp)
+    old_id = bson.ObjectId(now + b"\x00\x00\x00\x00\x00\x00\x00\x00")
+    new_id = bson.ObjectId(now + b"\x00\x00\x00\x00\x00\x00\x00\x01")
+    object_id_class = Mock()
+    object_id_class.from_datetime = bson.ObjectId.from_datetime
+
+    object_id_class.return_value = new_id
+    with patch("bson.ObjectId", object_id_class):
+        library.write(symbol, ts1)
+
+    library.snapshot('s1')
+
+    object_id_class.return_value = old_id
+    with patch("bson.ObjectId", object_id_class):
+        library.write(symbol, ts2)
+
+    now_dt = datetime.fromtimestamp(now_timestamp)
+    assert library.read_metadata(symbol, as_of=now_dt).version == 2
+
+
 def test_read_metadata_throws_on_deleted_symbol(library):
     library.write(symbol, ts1, metadata={'key': 'value'})
     library.delete(symbol)
@@ -948,6 +971,26 @@ def test_list_symbols_regex(library):
     assert library.list_symbols(b={'$gt': 5}, regex='asd') == ['asdf']
 
 
+def test_list_symbols_newer_version_with_lower_id(library):
+    now = struct.pack(">i", int(time.time()))
+    old_id = bson.ObjectId(now + b"\x00\x00\x00\x00\x00\x00\x00\x00")
+    new_id = bson.ObjectId(now + b"\x00\x00\x00\x00\x00\x00\x00\x01")
+    object_id_class = Mock()
+    object_id_class.from_datetime = bson.ObjectId.from_datetime
+
+    object_id_class.return_value = new_id
+    with patch("bson.ObjectId", object_id_class):
+        library.write(symbol, ts1)
+
+    library.snapshot('s1')
+
+    object_id_class.return_value = old_id
+    with patch("bson.ObjectId", object_id_class):
+        library.delete(symbol)
+
+    assert symbol not in library.list_symbols()
+
+
 def test_date_range_large(library):
     index = [dt(2017,1,1)]*20000 + [dt(2017,1,2)]*20000
     data = np.random.random((40000, 10))

tests/unit/store/test_version_store.py

@@ -105,7 +105,7 @@ def test_read_as_of_LondonTime():
     versions = vs._versions.with_options.return_value
     versions.find_one.assert_called_once_with({'symbol':'symbol', '_id':
                                               {'$lt': bson.ObjectId.from_datetime(dt(2013, 4, 1, 9, 0, tzinfo=mktz()) + dtd(seconds=1))}},
-                                             sort=[('_id', pymongo.DESCENDING)])
+                                             sort=[('symbol', pymongo.DESCENDING), ('version', pymongo.DESCENDING)])
 
 
 def test_read_as_of_NotNaive():
@@ -116,7 +116,7 @@ def test_read_as_of_NotNaive():
     versions = vs._versions.with_options.return_value
     versions.find_one.assert_called_once_with({'symbol':'symbol', '_id':
                                               {'$lt': bson.ObjectId.from_datetime(dt(2013, 4, 1, 9, 0, tzinfo=mktz('Europe/Paris')) + dtd(seconds=1))}},
-                                             sort=[('_id', pymongo.DESCENDING)])
+                                             sort=[('symbol', pymongo.DESCENDING), ('version', pymongo.DESCENDING)])
 
 
 def test_read_metadata_no_asof():
@@ -238,6 +238,22 @@ def test_snapshot():
                                                 call('foo', as_of=None, read_preference=ReadPreference.PRIMARY)])
 
 
+def test_list_symbols_default_pipeline():
+    versions = Mock()
+    vs = create_autospec(VersionStore, _versions=versions)
+    versions.aggregate.return_value = []
+
+    VersionStore.list_symbols(vs)
+
+    pipeline = [
+        {'$sort': bson.SON([('symbol', pymongo.DESCENDING), ('version', pymongo.DESCENDING)])},
+        {'$group': {'_id': '$symbol', 'deleted': {'$first': '$metadata.deleted'}}},
+        {'$match': {'deleted': {'$ne': True}}},
+        {'$project': {'_id': 0, 'symbol': '$_id'}}
+    ]
+    versions.aggregate.assert_called_once_with(pipeline)
+
+
 def test_snapshot_duplicate_raises_exception():
     vs = create_autospec(VersionStore, _snapshots=Mock())
     with pytest.raises(DuplicateSnapshotException) as e: