Merge pull request pandas-dev#715 from shashank88/remove_changes_coll

Fixes pandas-dev#714 Do not write to changes collection

Author: shashank88

arctic/fixtures/arctic.py

@@ -77,47 +77,6 @@ def multicolumn_store_with_uncompressed_write(mongo_server):
         }
     ])
     db.TEST.ARCTIC.update_one({"_id": "ARCTIC_META"}, {"$set": {"_id": "ARCTIC_META", "TYPE": "VersionStore", "QUOTA": 10737418240}})
-    db.TEST.changes.insert_many([
-        {
-            'append_count': 0,
-            'dtype_metadata': {
-                'index': ['index'],
-                'columns': ["('a', 'a')", "('w', 'v')"]
-            },
-            'segment_count': 1,
-            'dtype': '[(\'index\', \'S1\'), ("(\'a\', \'a\')", \'<i8\'), ("(\'w\', \'v\')", \'<i8\')]',
-            'symbol': 'pandas',
-            'up_to': 1,
-            'metadata': None,
-            'sha': bson.Binary(b'\xf2\x15h\x9d\x925\x95\xa5\x0e\x95J\xc4x\xfc\xfc\xd5\x80\xe0\x1d\xef', 0),
-            'shape': [-1],
-            'version': 1,
-            'base_sha': bson.Binary(b'\xf2\x15h\x9d\x925\x95\xa5\x0e\x95J\xc4x\xfc\xfc\xd5\x80\xe0\x1d\xef', 0),
-            '_id': bson.ObjectId('5ad0dc065c911d1188b512d8'),
-            'type': 'pandasdf',
-            'append_size': 0
-        },
-        {
-            'append_count': 1,
-            'dtype_metadata': {
-                'index': ['index'],
-                'columns': ["('a', 'a')", "('w', 'v')"]
-            },
-            'segment_count': 2,
-            'sha': bson.Binary(b'1\x83[ZO\xec\x080D\x80f\xe4@\xe4\xd3\x94yG\xe2\x08', 0),
-            'dtype': '[(\'index\', \'S1\'), ("(\'a\', \'a\')", \'<i8\'), ("(\'w\', \'v\')", \'<i8\')]',
-            'symbol': 'pandas',
-            'up_to': 2,
-            'metadata': None,
-            'base_version_id': bson.ObjectId('5ad0dc065c911d1188b512d8'),
-            'shape': [-1],
-            'version': 2,
-            'base_sha': bson.Binary(b'\xf2\x15h\x9d\x925\x95\xa5\x0e\x95J\xc4x\xfc\xfc\xd5\x80\xe0\x1d\xef', 0),
-            '_id': bson.ObjectId('5ad0dc075c911d1188b512d9'),
-            'type': 'pandasdf',
-            'append_size': 17
-        }
-    ])
     db.TEST.version_nums.insert_one({'symbol': 'pandas', '_id': bson.ObjectId('5ad0dc067934ecad404070bd'), 'version': 2})
     db.TEST.versions.insert_many([
         {
@@ -206,41 +165,6 @@ def ndarray_store_with_uncompressed_write(mongo_server):
         }
     ])
     db.TEST.ARCTIC.update_one({"_id": "ARCTIC_META"}, {"$set": {"_id": "ARCTIC_META", "TYPE": "VersionStore", "QUOTA": 10737418240}})
-    db.TEST.changes.insert_many([
-        {
-            "_id": bson.ObjectId("5ad0742c5c911d4d80ee2ea3"),
-            "append_count": 0,
-            "dtype_metadata": {},
-            "segment_count": 1,
-            "dtype": "[('abc', '<i8')]",
-            "symbol": "MYARR",
-            "up_to": 1,
-            "append_size": 0,
-            "sha": bson.Binary(base64.b64decode("Bf5AV1MWbxJVWefJrFWGVPEHx+k="), 0),
-            "shape": [-1],
-            "version": 1,
-            "base_sha": bson.Binary(base64.b64decode("Bf5AV1MWbxJVWefJrFWGVPEHx+k="), 0),
-            "type": "ndarray",
-            "metadata": None
-        },
-        {
-            "_id": bson.ObjectId("5ad0742c5c911d4d80ee2ea4"),
-            "append_count": 1,
-            "dtype_metadata": {},
-            "segment_count": 2,
-            "base_version_id": bson.ObjectId("5ad0742c5c911d4d80ee2ea3"),
-            "dtype": "[('abc', '<i8')]",
-            "symbol": "MYARR",
-            "up_to": 2,
-            "append_size": 8,
-            "sha": bson.Binary(base64.b64decode("Ax7oBxVFw1/9wKog2gfOLjbOVD8="), 0),
-            "shape": [-1],
-            "version": 2,
-            "base_sha": bson.Binary(base64.b64decode("Bf5AV1MWbxJVWefJrFWGVPEHx+k="), 0),
-            "type": "ndarray",
-            "metadata": None
-        }
-    ])
     db.TEST.versions_nums.insert_one({"_id": bson.ObjectId("5ad0742ca0949de6727cf993"), "symbol": "MYARR", "version": 2})
     db.TEST.versions.insert_many([
         {

arctic/store/version_store.py

@@ -50,10 +50,6 @@ class VersionStore(object):
     def initialize_library(cls, arctic_lib, hashed=True, **kwargs):
         c = arctic_lib.get_top_level_collection()
 
-        if '%s.changes' % c.name not in mongo_retry(c.database.list_collection_names)():
-            # 32MB buffer for change notifications
-            mongo_retry(c.database.create_collection)('%s.changes' % c.name, capped=True, size=32 * 1024 * 1024)
-
         if 'strict_write_handler' in kwargs:
             arctic_lib.set_library_metadata('STRICT_WRITE_HANDLER_MATCH',
                                             bool(kwargs.pop('strict_write_handler')))
@@ -109,9 +105,6 @@ def _reset(self):
         self._snapshots = self._collection.snapshots
         self._versions = self._collection.versions
         self._version_nums = self._collection.version_nums
-        self._publish_changes = '%s.changes' % self._collection.name in self._collection.database.list_collection_names()
-        if self._publish_changes:
-            self._changes = self._collection.changes
 
     def __getstate__(self):
         return {'arctic_lib': self._arctic_lib}
@@ -611,8 +604,6 @@ def append(self, symbol, data, metadata=None, prune_previous_version=True, upser
         else:
             raise Exception("Append not implemented for handler %s" % handler)
 
-        self._publish_change(symbol, version)
-
         if prune_previous_version and previous_version:
             # Does not allow prune to remove the base of the new version
             self._prune_previous_versions(symbol, keep_version=version.get('base_version_id'),
@@ -626,10 +617,6 @@ def append(self, symbol, data, metadata=None, prune_previous_version=True, upser
                              metadata=version.pop('metadata', None), data=None,
                              host=self._arctic_lib.arctic.mongo_host)
 
-    def _publish_change(self, symbol, version):
-        if self._publish_changes:
-            mongo_retry(self._changes.insert_one)(version)
-
     @mongo_retry
     def write(self, symbol, data, metadata=None, prune_previous_version=True, **kwargs):
         """
@@ -673,8 +660,6 @@ def write(self, symbol, data, metadata=None, prune_previous_version=True, **kwar
         if prune_previous_version and previous_version:
             self._prune_previous_versions(symbol, new_version_shas=version.get(FW_POINTERS_REFS_KEY))
 
-        self._publish_change(symbol, version)
-
         # Insert the new version into the version DB
         self._insert_version(version)
 
@@ -723,8 +708,6 @@ def _add_new_version_using_reference(self, symbol, new_version, reference_versio
 
         logger.debug('Finished updating versions with new metadata for %s', symbol)
 
-        self._publish_change(symbol, new_version)
-
         return VersionedItem(symbol=symbol, library=self._arctic_lib.get_name(), version=new_version['version'],
                              metadata=new_version.get('metadata'), data=None,
                              host=self._arctic_lib.arctic.mongo_host)

tests/integration/store/test_version_store.py

@@ -1564,24 +1564,6 @@ def test_append_does_not_duplicate_data_when_prune_fails(library, fw_pointers_cf
         assert len(set(data.index)) == len(data.index)
 
 
-@pytest.mark.parametrize('fw_pointers_cfg', [FwPointersCfg.DISABLED, FwPointersCfg.HYBRID, FwPointersCfg.ENABLED])
-def test_append_does_not_duplicate_data_when_publish_fails(library, fw_pointers_cfg):
-    with FwPointersCtx(fw_pointers_cfg):
-        side_effect = [OperationFailure(0), arctic.store.version_store.VersionStore._publish_change]
-        new_data = read_str_as_pandas("""times | near
-        2013-01-01 17:06:11.040 |  7.0
-        2013-01-02 17:06:11.040 |  8.2
-        2013-01-03 17:06:11.040 |  3.5
-        2013-01-04 17:06:11.040 |  0.7""")
-        library.write(symbol, ts1)
-
-        with patch.object(arctic.store.version_store.VersionStore, "_publish_change", autospec=True, side_effect=side_effect):
-            library.append(symbol, new_data)
-
-        data = library.read(symbol).data
-        assert len(set(data.index)) == len(data.index)
-
-
 @pytest.mark.parametrize('fw_pointers_cfg', [FwPointersCfg.DISABLED, FwPointersCfg.HYBRID, FwPointersCfg.ENABLED])
 def test_write_does_not_succeed_with_a_prune_error(library, fw_pointers_cfg):
     with FwPointersCtx(fw_pointers_cfg):
@@ -1596,19 +1578,6 @@ def test_write_does_not_succeed_with_a_prune_error(library, fw_pointers_cfg):
         assert len(library.list_versions(symbol)) == 1
 
 
-@pytest.mark.parametrize('fw_pointers_cfg', [FwPointersCfg.DISABLED, FwPointersCfg.HYBRID, FwPointersCfg.ENABLED])
-def test_write_does_not_succeed_with_a_publish_error(library, fw_pointers_cfg):
-    with FwPointersCtx(fw_pointers_cfg):
-        # More than max retries OperationFailure would be more realistic, but ValueError is used for simplicity
-        side_effect = [ValueError, arctic.store.version_store.VersionStore._publish_change]
-
-        with patch.object(arctic.store.version_store.VersionStore, "_publish_change", autospec=True, side_effect=side_effect):
-            with pytest.raises(ValueError):
-                library.append(symbol, ts1)
-
-        assert not library.list_versions(symbol)
-
-
 @pytest.mark.parametrize('fw_pointers_cfg', [FwPointersCfg.DISABLED, FwPointersCfg.HYBRID, FwPointersCfg.ENABLED])
 def test_prune_keeps_version(library, fw_pointers_cfg):
     with FwPointersCtx(fw_pointers_cfg):

tests/unit/store/test_version_store.py

@@ -159,8 +159,7 @@ def test_write_check_quota():
                      _version_nums=Mock(find_one_and_update=Mock(return_value={'version': 1})),
                      _versions=Mock(insert_one=lambda x: None),
                      _arctic_lib=create_autospec(ArcticLibraryBinding,
-                                                 arctic=create_autospec(Arctic, mongo_host='some_host')),
-                     _publish_changes=False)
+                                                 arctic=create_autospec(Arctic, mongo_host='some_host')))
     vs._collection.database.connection.nodes = []
     vs._write_handler.return_value = write_handler
     VersionStore.write(vs, 'sym', sentinel.data, prune_previous_version=False)
@@ -367,7 +366,6 @@ def test_write_metadata_with_previous_data():
         assert expected_ret_val == VersionStore.write_metadata(vs, symbol=TEST_SYMBOL, metadata=META_TO_WRITE)
         assert vs._versions.insert_one.call_args_list == [call(expected_new_version)]
         assert vs._versions.delete_one.called is False
-        assert vs._publish_change.call_args_list == [call(TEST_SYMBOL, expected_new_version)]
         assert vs.write.called is False
 
 
@@ -393,7 +391,6 @@ def test_write_empty_metadata():
         assert expected_ret_val == VersionStore.write_metadata(vs, symbol=TEST_SYMBOL, metadata=None)
         assert vs._versions.insert_one.call_args_list == [call(expected_new_version)]
         assert vs._versions.delete_one.called is False
-        assert vs._publish_change.call_args_list == [call(TEST_SYMBOL, expected_new_version)]
         assert vs.write.called is False
 
 
@@ -404,7 +401,6 @@ def test_write_metadata_insert_version_dupkeyerror():
     VersionStore.write_metadata(vs, symbol=TEST_SYMBOL, metadata=META_TO_WRITE)
     assert vs._version_nums.find_one_and_update.call_count == 2
     assert vs._versions.insert_one.call_count == 2
-    assert vs._publish_change.call_count == 1
 
 
 def test_write_metadata_insert_version_opfailure():
@@ -414,7 +410,6 @@ def test_write_metadata_insert_version_opfailure():
     VersionStore.write_metadata(vs, symbol=TEST_SYMBOL, metadata=META_TO_WRITE)
     assert vs._version_nums.find_one_and_update.call_count == 1
     assert vs._versions.insert_one.call_count == 2
-    assert vs._publish_change.call_count == 1
 
 
 def test_restore_version():
@@ -455,7 +450,6 @@ def test_restore_version_data_missing_symbol():
                                          as_of=TPL_VERSION['version'], prune_previous_version=True)
     assert vs._read_metadata.call_args_list == [call(TEST_SYMBOL, as_of=TPL_VERSION['version'])]
     assert vs._versions.insert_one.called is False
-    assert vs._publish_change.called is False
 
 
 def test_restore_last_version():
@@ -488,8 +482,7 @@ def test_write_error_clean_retry():
                          _version_nums=Mock(find_one_and_update=Mock(return_value={'version': 1})),
                          _versions=Mock(insert_one=Mock(__name__="insert_one"), find_one=Mock(__name__="find_one")),
                          _arctic_lib=create_autospec(ArcticLibraryBinding,
-                                                     arctic=create_autospec(Arctic, mongo_host='some_host')),
-                         _publish_changes=False)
+                                                     arctic=create_autospec(Arctic, mongo_host='some_host')))
     vs._insert_version = lambda version: VersionStore._insert_version(vs, version)
     vs._collection.database.connection.nodes = []
     vs._write_handler.return_value = write_handler
@@ -498,7 +491,6 @@ def test_write_error_clean_retry():
     assert vs._versions.find_one.call_count == 2
     assert write_handler.write.call_count == 2
     assert vs._versions.insert_one.call_count == 1
-    assert vs._publish_change.call_count == 1
 
 
 def test_write_insert_version_duplicatekey():
@@ -508,8 +500,7 @@ def test_write_insert_version_duplicatekey():
                          _version_nums=Mock(find_one_and_update=Mock(return_value={'version': 1})),
                          _versions=Mock(insert_one=Mock(__name__="insert_one"), find_one=Mock(__name__="find_one")),
                          _arctic_lib=create_autospec(ArcticLibraryBinding,
-                                                     arctic=create_autospec(Arctic, mongo_host='some_host')),
-                         _publish_changes=False)
+                                                     arctic=create_autospec(Arctic, mongo_host='some_host')))
     vs._insert_version = lambda version: VersionStore._insert_version(vs, version)
     vs._versions.insert_one.side_effect = [DuplicateKeyError("dup key error"), None]
     vs._collection.database.connection.nodes = []
@@ -518,7 +509,6 @@ def test_write_insert_version_duplicatekey():
     assert vs._version_nums.find_one_and_update.call_count == 2
     assert vs._versions.find_one.call_count == 2
     assert write_handler.write.call_count == 2
-    assert vs._publish_change.call_count == 2
     assert vs._versions.insert_one.call_count == 2
 
 
@@ -529,8 +519,7 @@ def test_write_insert_version_operror():
                          _version_nums=Mock(find_one_and_update=Mock(return_value={'version': 1})),
                          _versions=Mock(insert_one=Mock(__name__="insert_one"), find_one=Mock(__name__="find_one")),
                          _arctic_lib=create_autospec(ArcticLibraryBinding,
-                                                     arctic=create_autospec(Arctic, mongo_host='some_host')),
-                         _publish_changes=False)
+                                                     arctic=create_autospec(Arctic, mongo_host='some_host')))
     vs._insert_version = lambda version: VersionStore._insert_version(vs, version)
     vs._versions.insert_one.side_effect = [OperationFailure("mongo op error"), None]
     vs._collection.database.connection.nodes = []
@@ -540,7 +529,6 @@ def test_write_insert_version_operror():
     assert vs._versions.find_one.call_count == 1
     assert write_handler.write.call_count == 1
     assert vs._versions.insert_one.call_count == 2
-    assert vs._publish_change.call_count == 1
 
 
 def test_append_error_clean_retry():
@@ -553,8 +541,7 @@ def test_append_error_clean_retry():
                          _version_nums=Mock(find_one_and_update=Mock(return_value={'version': previous_version['version']+1})),
                          _versions=Mock(insert_one=Mock(__name__="insert_one"), find_one=Mock(__name__="find_one", return_value=previous_version)),
                          _arctic_lib=create_autospec(ArcticLibraryBinding,
-                                                     arctic=create_autospec(Arctic, mongo_host='some_host')),
-                         _publish_changes=False)
+                                                     arctic=create_autospec(Arctic, mongo_host='some_host')))
     vs._insert_version = lambda version: VersionStore._insert_version(vs, version)
     vs._collection.database.connection.nodes = []
     vs._read_handler.return_value = read_handler
@@ -563,7 +550,6 @@ def test_append_error_clean_retry():
     assert vs._versions.find_one.call_count == 2
     assert read_handler.append.call_count == 2
     assert vs._versions.insert_one.call_count == 1
-    assert vs._publish_change.call_count == 1
 
 
 def test_append_insert_version_duplicatekey():
@@ -575,8 +561,7 @@ def test_append_insert_version_duplicatekey():
                          _version_nums=Mock(find_one_and_update=Mock(return_value={'version': previous_version['version']+1})),
                          _versions=Mock(insert_one=Mock(__name__="insert_one"), find_one=Mock(__name__="find_one", return_value=previous_version)),
                          _arctic_lib=create_autospec(ArcticLibraryBinding,
-                                                     arctic=create_autospec(Arctic, mongo_host='some_host')),
-                         _publish_changes=False)
+                                                     arctic=create_autospec(Arctic, mongo_host='some_host')))
     vs._insert_version = lambda version: VersionStore._insert_version(vs, version)
     vs._versions.insert_one.side_effect = [DuplicateKeyError("dup key error"), None]
     vs._collection.database.connection.nodes = []
@@ -586,7 +571,6 @@ def test_append_insert_version_duplicatekey():
     assert vs._versions.find_one.call_count == 2
     assert read_handler.append.call_count == 2
     assert vs._versions.insert_one.call_count == 2
-    assert vs._publish_change.call_count == 2
 
 def test_append_insert_version_operror():
     read_handler = Mock(append=Mock(__name__=""))
@@ -597,8 +581,7 @@ def test_append_insert_version_operror():
                          _version_nums=Mock(find_one_and_update=Mock(return_value={'version': previous_version['version']+1})),
                          _versions=Mock(insert_one=Mock(__name__="insert_one"), find_one=Mock(__name__="find_one", return_value=previous_version)),
                          _arctic_lib=create_autospec(ArcticLibraryBinding,
-                                                     arctic=create_autospec(Arctic, mongo_host='some_host')),
-                         _publish_changes=False)
+                                                     arctic=create_autospec(Arctic, mongo_host='some_host')))
     vs._insert_version = lambda version: VersionStore._insert_version(vs, version)
     vs._versions.insert_one.side_effect = [OperationFailure("mongo op error"), None]
     vs._collection.database.connection.nodes = []
@@ -608,4 +591,3 @@ def test_append_insert_version_operror():
     assert vs._versions.find_one.call_count == 1
     assert read_handler.append.call_count == 1
     assert vs._versions.insert_one.call_count == 2
-    assert vs._publish_change.call_count == 1