Added support in DB for is_wip flag to (hopefully) speed up cache updates.

Also fixed/improved setUp in integrations/server/test_covidcast_meta.py and added skeleton-code for new `missing_*` rows.

Author: melange396

integrations/acquisition/covidcast/test_covidcast_meta_caching.py

@@ -66,14 +66,14 @@ def test_caching(self):
     self.cur.execute('''
       insert into covidcast values
         (0, 'src', 'sig', 'day', 'state', 20200422, 'pa',
-          123, 1, 2, 3, 456, 1, 20200422, 0),
+          123, 1, 2, 3, 456, 1, 20200422, 0, False),
         (0, 'src', 'sig', 'day', 'state', 20200422, 'wa',
-          789, 1, 2, 3, 456, 1, 20200423, 1)
+          789, 1, 2, 3, 456, 1, 20200423, 1, False)
     ''')
     self.cur.execute('''
       insert into covidcast values
         (100, 'src', 'wip_sig', 'day', 'state', 20200422, 'pa',
-          456, 4, 5, 6, 789, -1, 20200422, 0)
+          456, 4, 5, 6, 789, -1, 20200422, 0, True)
     ''')
 
     self.cnx.commit()

integrations/acquisition/covidcast/test_direction_updating.py

@@ -61,23 +61,23 @@ def test_uploading(self):
     self.cur.execute('''
       insert into covidcast values
         (0, 'src', 'sig', 'day', 'state', 20200228, 'ca',
-          123, 2, 0, 0, 0, NULL, 20200228, 0),
+          123, 2, 0, 0, 0, NULL, 20200228, 0, False),
         (0, 'src', 'sig', 'day', 'state', 20200229, 'ca',
-          123, 6, 0, 0, 0, NULL, 20200229, 0),
+          123, 6, 0, 0, 0, NULL, 20200229, 0, False),
         (0, 'src', 'sig', 'day', 'state', 20200301, 'ca',
-          123, 5, 0, 0, 0, NULL, 20200301, 0),
+          123, 5, 0, 0, 0, NULL, 20200301, 0, False),
         (0, 'src', 'sig', 'day', 'state', 20200511, 'fl',
-          123, 1, 0, 0, 0, NULL, 20200511, 0),
+          123, 1, 0, 0, 0, NULL, 20200511, 0, False),
         (0, 'src', 'sig', 'day', 'state', 20200512, 'fl',
-          123, 2, 0, 0, 0, NULL, 20200512, 0),
+          123, 2, 0, 0, 0, NULL, 20200512, 0, False),
         (0, 'src', 'sig', 'day', 'state', 20200517, 'fl',
-          123, 2, 0, 0, 0, NULL, 20200517, 0),
+          123, 2, 0, 0, 0, NULL, 20200517, 0, False),
         (0, 'src', 'sig', 'day', 'state', 20200615, 'tx',
-          123, 9, 0, 0, 456, NULL, 20200615, 0),
+          123, 9, 0, 0, 456, NULL, 20200615, 0, False),
         (0, 'src', 'sig', 'day', 'state', 20200616, 'tx',
-          123, 5, 0, 0, 456, NULL, 20200616, 0),
+          123, 5, 0, 0, 456, NULL, 20200616, 0, False),
         (0, 'src', 'sig', 'day', 'state', 20200617, 'tx',
-          123, 1, 0, 0, 456, 1, 20200617, 0)
+          123, 1, 0, 0, 456, 1, 20200617, 0, False)
     ''')
     self.cnx.commit()
 

integrations/client/test_delphi_epidata.py

@@ -50,11 +50,11 @@ def test_covidcast(self):
     self.cur.execute('''
       insert into covidcast values
         (0, 'src', 'sig', 'day', 'county', 20200414, '01234',
-          123, 1.5, 2.5, 3.5, 456, 4, 20200414, 0),
+          123, 1.5, 2.5, 3.5, 456, 4, 20200414, 0, False),
         (0, 'src', 'sig', 'day', 'county', 20200414, '01234',
-          456, 5.5, 1.2, 10.5, 789, 0, 20200415, 1),
+          456, 5.5, 1.2, 10.5, 789, 0, 20200415, 1, False),
         (0, 'src', 'sig', 'day', 'county', 20200414, '01234',
-          345, 6.5, 2.2, 11.5, 678, 0, 20200416, 2)
+          345, 6.5, 2.2, 11.5, 678, 0, 20200416, 2, False)
     ''')
     self.cnx.commit()
 
@@ -157,11 +157,11 @@ def test_covidcast_meta(self):
     self.cur.execute('''
       insert into covidcast values
         (0, 'src', 'sig', 'day', 'county', 20200414, '01234',
-          123, 1.5, 2.5, 3.5, 456, 4, 20200414, 0),
+          123, 1.5, 2.5, 3.5, 456, 4, 20200414, 0, False),
         (0, 'src', 'sig', 'day', 'county', 20200414, '01234',
-          345, 6.0, 2.2, 11.5, 678, 0, 20200416, 2),
+          345, 6.0, 2.2, 11.5, 678, 0, 20200416, 2, False),
         (0, 'src', 'sig', 'day', 'county', 20200415, '01234',
-          345, 7.0, 2.0, 12.5, 678, 0, 20200416, 1)
+          345, 7.0, 2.0, 12.5, 678, 0, 20200416, 1, False)
     ''')
     self.cnx.commit()
 

integrations/server/test_covidcast.py

@@ -45,7 +45,7 @@ def test_round_trip(self):
     self.cur.execute('''
       insert into covidcast values
         (0, 'src', 'sig', 'day', 'county', 20200414, '01234',
-          123, 1.5, 2.5, 3.5, 456, 4, 20200414, 0)
+          123, 1.5, 2.5, 3.5, 456, 4, 20200414, 0, False)
     ''')
     self.cnx.commit()
 
@@ -85,17 +85,17 @@ def test_location_wildcard(self):
     self.cur.execute('''
       insert into covidcast values
         (0, 'src', 'sig', 'day', 'county', 20200414, '11111',
-          123, 10, 11, 12, 456, 13, 20200414, 0),
+          123, 10, 11, 12, 456, 13, 20200414, 0, False),
         (0, 'src', 'sig', 'day', 'county', 20200414, '22222',
-          123, 20, 21, 22, 456, 23, 20200414, 0),
+          123, 20, 21, 22, 456, 23, 20200414, 0, False),
         (0, 'src', 'sig', 'day', 'county', 20200414, '33333',
-          123, 30, 31, 32, 456, 33, 20200414, 0),
+          123, 30, 31, 32, 456, 33, 20200414, 0, False),
         (0, 'src', 'sig', 'day', 'msa', 20200414, '11111',
-          123, 40, 41, 42, 456, 43, 20200414, 0),
+          123, 40, 41, 42, 456, 43, 20200414, 0, False),
         (0, 'src', 'sig', 'day', 'msa', 20200414, '22222',
-          123, 50, 51, 52, 456, 53, 20200414, 0),
+          123, 50, 51, 52, 456, 53, 20200414, 0, False),
         (0, 'src', 'sig', 'day', 'msa', 20200414, '33333',
-          123, 60, 61, 62, 456, 634, 20200414, 0)
+          123, 60, 61, 62, 456, 634, 20200414, 0, False)
     ''')
     self.cnx.commit()
 
@@ -155,17 +155,17 @@ def test_location_timeline(self):
     self.cur.execute('''
       insert into covidcast values
         (0, 'src', 'sig', 'day', 'county', 20200411, '01234',
-          123, 10, 11, 12, 456, 13, 20200413, 2),
+          123, 10, 11, 12, 456, 13, 20200413, 2, False),
         (0, 'src', 'sig', 'day', 'county', 20200412, '01234',
-          123, 20, 21, 22, 456, 23, 20200413, 1),
+          123, 20, 21, 22, 456, 23, 20200413, 1, False),
         (0, 'src', 'sig', 'day', 'county', 20200413, '01234',
-          123, 30, 31, 32, 456, 33, 20200413, 0),
+          123, 30, 31, 32, 456, 33, 20200413, 0, False),
         (0, 'src', 'sig', 'day', 'county', 20200411, '11111',
-          123, 40, 41, 42, 456, 43, 20200413, 2),
+          123, 40, 41, 42, 456, 43, 20200413, 2, False),
         (0, 'src', 'sig', 'day', 'county', 20200412, '22222',
-          123, 50, 51, 52, 456, 53, 20200413, 1),
+          123, 50, 51, 52, 456, 53, 20200413, 1, False),
         (0, 'src', 'sig', 'day', 'county', 20200413, '33333',
-          123, 60, 61, 62, 456, 63, 20200413, 0)
+          123, 60, 61, 62, 456, 63, 20200413, 0, False)
     ''')
     self.cnx.commit()
 
@@ -225,7 +225,7 @@ def test_unique_key_constraint(self):
     self.cur.execute('''
       insert into covidcast values
         (0, 'src', 'sig', 'day', 'county', 20200414, '01234',
-          0, 0, 0, 0, 0, 0, 20200414, 0)
+          0, 0, 0, 0, 0, 0, 20200414, 0, False)
     ''')
     self.cnx.commit()
 
@@ -234,14 +234,14 @@ def test_unique_key_constraint(self):
       self.cur.execute('''
         insert into covidcast values
           (0, 'src', 'sig', 'day', 'county', 20200414, '01234',
-            1, 1, 1, 1, 1, 1, 20200414, 0)
+            1, 1, 1, 1, 1, 1, 20200414, 0, False)
       ''')
 
     # succeed to insert different dummy data under a different issue
     self.cur.execute('''
       insert into covidcast values
         (0, 'src', 'sig', 'day', 'county', 20200414, '01234',
-          1, 1, 1, 1, 1, 1, 20200415, 1)
+          1, 1, 1, 1, 1, 1, 20200415, 1, False)
     ''')
 
   def test_nullable_columns(self):
@@ -251,7 +251,7 @@ def test_nullable_columns(self):
     self.cur.execute('''
       insert into covidcast values
         (0, 'src', 'sig', 'day', 'county', 20200414, '01234',
-          123, 0.123, NULL, NULL, 456, NULL, 20200414, 0)
+          123, 0.123, NULL, NULL, 456, NULL, 20200414, 0, False)
     ''')
     self.cnx.commit()
 
@@ -291,15 +291,15 @@ def test_temporal_partitioning(self):
     self.cur.execute('''
       insert into covidcast values
         (0, 'src', 'sig', 'hour', 'state', 2020041714, 'vi',
-          123, 10, 11, 12, 456, 13, 2020041714, 0),
+          123, 10, 11, 12, 456, 13, 2020041714, 0, False),
         (0, 'src', 'sig', 'day', 'state', 20200417, 'vi',
-          123, 20, 21, 22, 456, 23, 20200417, 00),
+          123, 20, 21, 22, 456, 23, 20200417, 00, False),
         (0, 'src', 'sig', 'week', 'state', 202016, 'vi',
-          123, 30, 31, 32, 456, 33, 202016, 0),
+          123, 30, 31, 32, 456, 33, 202016, 0, False),
         (0, 'src', 'sig', 'month', 'state', 202004, 'vi',
-          123, 40, 41, 42, 456, 43, 202004, 0),
+          123, 40, 41, 42, 456, 43, 202004, 0, False),
         (0, 'src', 'sig', 'year', 'state', 2020, 'vi',
-          123, 50, 51, 52, 456, 53, 2020, 0)
+          123, 50, 51, 52, 456, 53, 2020, 0, False)
     ''')
     self.cnx.commit()
 

integrations/server/test_covidcast_meta.py

@@ -28,6 +28,7 @@ def setUp(self):
         database='epidata')
     cur = cnx.cursor()
     cur.execute('truncate table covidcast')
+    cur.execute('update covidcast_meta_cache set timestamp = 0, epidata = ""')
     cnx.commit()
     cur.close()
 
@@ -46,7 +47,7 @@ def test_round_trip(self):
     # insert dummy data and accumulate expected results (in sort order)
     template = '''
       insert into covidcast values
-        (0, "%s", "%s", "%s", "%s", %d, "%s", 123, %d, 0, 0, 456, 0, %d, 0)
+        (0, "%s", "%s", "%s", "%s", %d, "%s", 123, %d, 0, 0, 456, 0, %d, 0, %d)
     '''
     expected = []
     for src in ('src1', 'src2'):
@@ -72,7 +73,7 @@ def test_round_trip(self):
             })
             for tv in (1, 2):
               for gv, v in zip(('geo1', 'geo2'), (10, 20)):
-                self.cur.execute(template % (src, sig, tt, gt, tv, gv, v, tv))
+                self.cur.execute(template % (src, sig, tt, gt, tv, gv, v, tv, False))
     self.cnx.commit()
     update_cache(args=None)
 
@@ -94,14 +95,18 @@ def test_suppress_work_in_progress(self):
     # insert dummy data and accumulate expected results (in sort order)
     template = '''
       insert into covidcast values
-        (0, "%s", "%s", "%s", "%s", %d, "%s", 123, %d, 0, 0, 456, 0, %d, 0)
+        (0, "%s", "%s", "%s", "%s", %d, "%s", 123, %d, 0, 0, 456, 0, %d, 0, %d)
     '''
     expected = []
     for src in ('src1', 'src2'):
       for sig in ('sig1', 'sig2', 'wip_sig3'):
         for tt in ('day', 'week'):
           for gt in ('hrr', 'msa'):
-            if sig != 'wip_sig3':
+
+            if sig == 'wip_sig3':
+              is_wip = True
+            else:
+              is_wip = False
               expected.append({
                 'data_source': src,
                 'signal': sig,
@@ -121,7 +126,7 @@ def test_suppress_work_in_progress(self):
               })
             for tv in (1, 2):
               for gv, v in zip(('geo1', 'geo2'), (10, 20)):
-                self.cur.execute(template % (src, sig, tt, gt, tv, gv, v, tv))
+                self.cur.execute(template % (src, sig, tt, gt, tv, gv, v, tv, is_wip))
     self.cnx.commit()
     update_cache(args=None)
 

src/acquisition/covidcast/csv_to_database.py

@@ -74,11 +74,16 @@ def archive_as_successful(path_src, filename, source):
 
     (source, signal, time_type, geo_type, time_value, issue, lag) = details
 
+    is_wip = False
+    if signal[:4].lower() == "wip_":
+      is_wip = True
+    print(signal, is_wip)
+
     csv_rows = csv_importer_impl.load_csv(path, geo_type)
 
     all_rows_valid = False
     try:
-      cc_rows = CovidcastRow.fromCsvRows(csv_rows, source, signal, time_type, geo_type, time_value, issue, lag)
+      cc_rows = CovidcastRow.fromCsvRows(csv_rows, source, signal, time_type, geo_type, time_value, issue, lag, is_wip)
       rows_list = list(cc_rows)
       if not rows_list:
         raise ValueError("No data")

src/acquisition/covidcast/database.py

@@ -17,21 +17,21 @@ class CovidcastRow():
   """A container for all the values of a single covidcast row."""
 
   @staticmethod
-  def fromCsvRowValue(row_value, source, signal, time_type, geo_type, time_value, issue, lag):
+  def fromCsvRowValue(row_value, source, signal, time_type, geo_type, time_value, issue, lag, is_wip):
     return CovidcastRow(source, signal, time_type, geo_type, time_value,
                         row_value.geo_value,
                         row_value.value,
                         row_value.stderr,
                         row_value.sample_size,
-                        issue, lag)
+                        issue, lag, is_wip)
 
   @staticmethod
-  def fromCsvRows(row_values, source, signal, time_type, geo_type, time_value, issue, lag):
+  def fromCsvRows(row_values, source, signal, time_type, geo_type, time_value, issue, lag, is_wip):
     # NOTE: returns a generator, as row_values is expected to be a generator
-    return (CovidcastRow.fromCsvRowValue(row_value, source, signal, time_type, geo_type, time_value, issue, lag)
+    return (CovidcastRow.fromCsvRowValue(row_value, source, signal, time_type, geo_type, time_value, issue, lag, is_wip)
             for row_value in row_values)
 
-  def __init__(self, source, signal, time_type, geo_type, time_value, geo_value, value, stderr, sample_size, issue, lag):
+  def __init__(self, source, signal, time_type, geo_type, time_value, geo_value, value, stderr, sample_size, issue, lag, is_wip):
     self.id = None
     self.source = source
     self.signal = signal
@@ -46,6 +46,7 @@ def __init__(self, source, signal, time_type, geo_type, time_value, geo_value, v
     self.direction = None
     self.issue = issue
     self.lag = lag
+    self.is_wip = is_wip
 
 
 class Database:
@@ -103,12 +104,12 @@ def insert_or_update_batch(self, cc_rows, batch_size=0, commit_partial=False):
         (`id`, `source`, `signal`, `time_type`, `geo_type`, `time_value`, `geo_value`,
         `value_updated_timestamp`, `value`, `stderr`, `sample_size`,
         `direction_updated_timestamp`, `direction`,
-        `issue`, `lag`)
+        `issue`, `lag`, `is_wip`)
       VALUES
         (0, %s, %s, %s, %s, %s, %s,
         UNIX_TIMESTAMP(NOW()), %s, %s, %s,
         0, NULL,
-        %s, %s)
+        %s, %s, %s)
       ON DUPLICATE KEY UPDATE
         `value_updated_timestamp` = VALUES(`value_updated_timestamp`),
         `value` = VALUES(`value`),
@@ -139,7 +140,8 @@ def insert_or_update_batch(self, cc_rows, batch_size=0, commit_partial=False):
         row.stderr,
         row.sample_size,
         row.issue,
-        row.lag
+        row.lag,
+        row.is_wip
       ) for row in cc_rows[start:end]]
 
       result = self._cursor.executemany(sql, args)
@@ -164,7 +166,8 @@ def insert_or_update(
       stderr,
       sample_size,
       issue,
-      lag):
+      lag,
+      is_wip):
     """
     Insert a new row, or update an existing row, in the `covidcast` table.
 
@@ -173,7 +176,7 @@ def insert_or_update(
 
     sql = '''
       INSERT INTO `covidcast` VALUES
-        (0, %s, %s, %s, %s, %s, %s, UNIX_TIMESTAMP(NOW()), %s, %s, %s, 0, NULL, %s, %s)
+        (0, %s, %s, %s, %s, %s, %s, UNIX_TIMESTAMP(NOW()), %s, %s, %s, 0, NULL, %s, %s, %s)
       ON DUPLICATE KEY UPDATE
         `value_updated_timestamp` = VALUES(`value_updated_timestamp`),
         `value` = VALUES(`value`),
@@ -192,7 +195,8 @@ def insert_or_update(
       stderr,
       sample_size,
       issue,
-      lag
+      lag,
+      is_wip
     )
 
     self._cursor.execute(sql, args)
@@ -587,7 +591,7 @@ def get_covidcast_meta(self):
           x.`geo_type` = t.`geo_type` AND
           x.`geo_value` = t.`geo_value`
       WHERE
-        t.`signal` NOT LIKE 'wip_%'
+        NOT t.`is_wip`
       GROUP BY
         t.`source`,
         t.`signal`,