Add index for track owner id and udpate plays mat view to concurrent (#…

…865)

* Add index for track owner id and udpate plays mat view to concurrent

* Increase index_plays task interval

discovery-provider/alembic/versions/281de8af4b93_add_indexes.py

@@ -0,0 +1,41 @@
+"""add indexes
+
+Revision ID: 281de8af4b93
+Revises: ffcb2df7b0ee
+Create Date: 2020-09-28 15:27:40.389787
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = '281de8af4b93'
+down_revision = 'ffcb2df7b0ee'
+branch_labels = None
+depends_on = None
+
+def upgrade():
+    # Add an index for track owner id on the tracks table
+    op.create_index(op.f('track_owner_id_idx'),
+                'tracks', ['owner_id'], unique=False)
+
+    # Update the index on the aggregate_plays materialized view to be unique for concurrent updates
+    connection = op.get_bind()
+    connection.execute('''
+        DROP INDEX play_item_id_idx;
+        CREATE UNIQUE INDEX play_item_id_idx ON aggregate_plays (play_item_id);
+    ''')
+
+def downgrade():
+    # Drop the index for track owner id
+    op.drop_index(op.f('track_owner_id_idx'),
+                  table_name='tracks')
+    connection = op.get_bind()
+
+    # Update the index on the aggregate_plays materialized view to not be unique
+    connection.execute('''
+        DROP INDEX play_item_id_idx;
+        CREATE INDEX play_item_id_idx ON aggregate_plays (play_item_id);
+    ''')
+

discovery-provider/src/__init__.py

@@ -293,7 +293,7 @@ def configure_celery(flask_app, celery, test_config=None):
             },
             "update_play_count": {
                 "task": "update_play_count",
-                "schedule": timedelta(seconds=10)
+                "schedule": timedelta(seconds=60)
             },
             "update_metrics": {
                 "task": "update_metrics",

discovery-provider/src/tasks/index_plays.py

@@ -18,7 +18,7 @@
 
 def get_time_diff(previous_time):
     # Returns the time difference in milliseconds
-    return int(time.time() - previous_time * 1000)
+    return int((time.time() - previous_time) * 1000)
 
 # Retrieve the play counts from the identity service
 # NOTE: indexing the plays will eventually be a part of `index_blocks`
@@ -43,6 +43,8 @@ def get_track_plays(self, db):
         else:
             most_recent_play_date = most_recent_play_date[0].timestamp()
 
+        job_extra_info['most_recent_play_date'] = get_time_diff(start_time)
+
         # Create and query identity service endpoint for track play counts
         identity_url = update_play_count.shared_config['discprov']['identity_service_url']
         params = {'startTime': most_recent_play_date,
@@ -187,7 +189,7 @@ def get_track_plays(self, db):
 
         if plays:
             session.bulk_save_objects(plays)
-            session.execute("REFRESH MATERIALIZED VIEW aggregate_plays")
+            session.execute("REFRESH MATERIALIZED VIEW CONCURRENTLY aggregate_plays")
 
         job_extra_info['number_rows_insert'] = len(plays)
         job_extra_info['insert_refresh_time'] = get_time_diff(