adding SQL from Joe's dbjobs and an extrememely basic runner module

also did a bunch of formatting of said SQL code

Author: melange396

src/acquisition/covidcast/database.py

@@ -163,7 +163,9 @@ def insert_or_update_batch(self, cc_rows, batch_size=2**20, commit_partial=False
     '''
 
     update_status_sql = f'''
-        UPDATE `{self.load_table}` SET `process_status` = '{self.process_status['loaded']}' WHERE `process_status` = '{self.process_status['inserting']}'
+        UPDATE `{self.load_table}`
+            SET `process_status` = '{self.process_status['loaded']}'
+            WHERE `process_status` = '{self.process_status['inserting']}'
     '''
 
     if 0 != self.count_insertstatus_rows():
@@ -219,6 +221,110 @@ def insert_or_update_batch(self, cc_rows, batch_size=2**20, commit_partial=False
       raise e
     return total
 
+  def run_dbjobs(self):
+
+    signal_load_set_comp_keys = f'''
+        UPDATE `{self.load_table}`
+        SET compressed_signal_key = md5(CONCAT(`source`,`signal`)),
+            compressed_geo_key = md5(CONCAT(`geo_type`,`geo_value`))
+    '''
+
+    signal_load_mark_batch = f'''
+        UPDATE `{self.load_table}` 
+        SET process_status = 'b'
+    '''
+
+    signal_dim_add_new_load = f'''
+        INSERT INTO signal_dim (`source`, `signal`, `compressed_signal_key`) 
+            SELECT DISTINCT `source`, `signal`, compressed_signal_key 
+                FROM `{self.load_table}` 
+                WHERE compressed_signal_key NOT IN 
+                    (SELECT DISTINCT compressed_signal_key 
+                     FROM signal_dim)
+    '''
+
+    geo_dim_add_new_load = f'''
+        INSERT INTO geo_dim (`geo_type`, `geo_value`, `compressed_geo_key`) 
+            SELECT DISTINCT `geo_type`, `geo_value`, compressed_geo_key 
+                FROM `{self.load_table}` 
+                WHERE compressed_geo_key NOT IN 
+                    (SELECT DISTINCT compressed_geo_key 
+                     FROM geo_dim)
+    '''
+
+    signal_history_load = f'''
+        INSERT INTO signal_history 
+            (signal_data_id, signal_key_id, geo_key_id, demog_key_id, issue, data_as_of_dt, 
+             time_type, time_value, `value`, stderr, sample_size, `lag`, value_updated_timestamp, 
+             computation_as_of_dt, is_latest_issue, missing_value, missing_stderr, missing_sample_size, `id`)
+        SELECT
+            signal_data_id, sd.signal_key_id, gd.geo_key_id, 0, issue, data_as_of_dt, 
+                time_type, time_value, `value`, stderr, sample_size, `lag`, value_updated_timestamp, 
+                computation_as_of_dt, is_latest_issue, missing_value, missing_stderr, missing_sample_size, `id`
+            FROM `{self.load_table}` sl
+                INNER JOIN signal_dim sd
+                    USE INDEX(`compressed_signal_key_ind`)
+                    ON sd.compressed_signal_key = sl.compressed_signal_key
+                INNER JOIN geo_dim gd
+                    USE INDEX(`compressed_geo_key_ind`)
+                    ON gd.compressed_geo_key = sl.compressed_geo_key
+            WHERE process_status = 'b'
+        ON DUPLICATE KEY UPDATE
+            `value_updated_timestamp` = sl.`value_updated_timestamp`,
+            `value` = sl.`value`,
+            `stderr` = sl.`stderr`,
+            `sample_size` = sl.`sample_size`,
+            `lag` = sl.`lag`,
+            `missing_value` = sl.`missing_value`,
+            `missing_stderr` = sl.`missing_stderr`,
+            `missing_sample_size` = sl.`missing_sample_size`
+    '''
+
+    signal_latest_load = f'''
+        INSERT INTO signal_latest 
+            (signal_data_id, signal_key_id, geo_key_id, demog_key_id, issue, data_as_of_dt, 
+             time_type, time_value, `value`, stderr, sample_size, `lag`, value_updated_timestamp, 
+             computation_as_of_dt, missing_value, missing_stderr, missing_sample_size)
+        SELECT
+            signal_data_id, sd.signal_key_id, gd.geo_key_id, 0, issue, data_as_of_dt, 
+                time_type, time_value, `value`, stderr, sample_size, `lag`, value_updated_timestamp, 
+                computation_as_of_dt, missing_value, missing_stderr, missing_sample_size
+            FROM `{self.load_table}` sl 
+                INNER JOIN signal_dim sd 
+                    USE INDEX(`compressed_signal_key_ind`) 
+                    ON sd.compressed_signal_key = sl.compressed_signal_key 
+                INNER JOIN geo_dim gd 
+                    USE INDEX(`compressed_geo_key_ind`) 
+                    ON gd.compressed_geo_key = sl.compressed_geo_key 
+            WHERE process_status = 'b'
+                AND is_latest_issue = 1
+        ON DUPLICATE KEY UPDATE
+            `value_updated_timestamp` = sl.`value_updated_timestamp`,
+            `value` = sl.`value`,
+            `stderr` = sl.`stderr`,
+            `sample_size` = sl.`sample_size`,
+            `lag` = sl.`lag`,
+            `missing_value` = sl.`missing_value`,
+            `missing_stderr` = sl.`missing_stderr`,
+            `missing_sample_size` = sl.`missing_sample_size` 
+    '''
+
+    signal_load_delete_processed = f'''
+        DELETE FROM `{self.load_table}` 
+        WHERE  process_status <> 'l'
+    '''
+
+    self._cursor.execute(signal_load_set_comp_keys)
+    self._cursor.execute(signal_load_mark_batch)
+    self._cursor.execute(signal_dim_add_new_load)
+    self._cursor.execute(geo_dim_add_new_load)
+    self._cursor.execute(signal_history_load)
+    self._cursor.execute(signal_latest_load)
+    self._cursor.execute(signal_load_delete_processed)
+
+    return self
+
+
   def compute_covidcast_meta(self, table_name=None):
     """Compute and return metadata on all COVIDcast signals."""
     logger = get_structured_logger("compute_covidcast_meta")

src/acquisition/covidcast/dbjobs_runner.py

@@ -0,0 +1,13 @@
+
+from delphi.epidata.acquisition.covidcast.database import Database
+
+def main():
+  database = Database()
+  database.connect()
+  try:
+    database.run_dbjobs()
+  finally:
+    database.disconnect(True)
+
+if __name__ == '__main__':
+  main()