Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

feat: add min_issue to covidcast meta data #236

Closed
wants to merge 7 commits into from
Closed

feat: add min_issue to covidcast meta data #236

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
f723beb
feat: add min_issue to covidcast meta data
sgratzl Oct 7, 2020
f177818
Merge remote-tracking branch 'origin/main' into sgratzl/min_issue
sgratzl Oct 12, 2020
6f01aa5
fix: update min issue computation
sgratzl Oct 12, 2020
b2fce76
fix: python client test
sgratzl Oct 12, 2020
a581f0b
refactor: simpliy lookup
sgratzl Oct 15, 2020
06aab57
fix: test
sgratzl Oct 15, 2020
a85adbb
Merge remote-tracking branch 'origin/main' into sgratzl/min_issue
sgratzl Oct 21, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions docs/api/covidcast_meta.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,7 @@ None required.
| `epidata[].max_value` | maximum value | float |
| `epidata[].mean_value` | mean of value | float |
| `epidata[].stdev_value` | standard deviation of value | float |
| `epidata[].min_issue` | earliest date data was issued (e.g., 20200710) | integer |
| `epidata[].max_issue` | most recent date data was issued (e.g., 20200710) | integer |
| `epidata[].min_lag` | smallest lag from observation to issue, in `time_type` units | integer |
| `epidata[].max_lag` | largest lag from observation to issue, in `time_type` units | integer |
Expand Down
5 changes: 3 additions & 2 deletions integrations/acquisition/covidcast/test_covidcast_meta_caching.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -97,8 +97,9 @@ def test_caching(self):
'last_update': 789,
'min_value': 1,
'max_value': 1,
'mean_value': 1,
'stdev_value': 0,
'mean_value': 1.0,
'stdev_value': 0.0,
'min_issue': 20200422,
'max_issue': 20200423,
'min_lag': 0,
'max_lag': 1,
Expand Down
1 change: 1 addition & 0 deletions integrations/client/test_delphi_epidata.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -355,6 +355,7 @@ def test_covidcast_meta(self):
'mean_value': 6.5,
'stdev_value': 0.5,
'last_update': 345,
'min_issue': 20200414,
'max_issue': 20200416,
'min_lag': 1,
'max_lag': 2,
Expand Down
58 changes: 57 additions & 1 deletion integrations/server/test_covidcast_meta.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,7 +47,11 @@ 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, 1, %d)
(0, "%s", "%s", "%s", "%s", %d, "%s",
123,
%d, 0, 0,
456, 0,
%d, 0, 1, %d)
'''
expected = []
for src in ('src1', 'src2'):
Expand All @@ -67,6 +71,7 @@ def test_round_trip(self):
'mean_value': 15,
'stdev_value': 5,
'last_update': 123,
'min_issue': 1,
'max_issue': 2,
'min_lag': 0,
'max_lag': 0,
Expand Down Expand Up @@ -116,6 +121,7 @@ def test_filter(self):
'mean_value': 15,
'stdev_value': 5,
'last_update': 123,
'min_issue': 1,
'max_issue': 2,
'min_lag': 0,
'max_lag': 0,
Expand Down Expand Up @@ -235,6 +241,7 @@ def test_suppress_work_in_progress(self):
'mean_value': 15,
'stdev_value': 5,
'last_update': 123,
'min_issue': 1,
'max_issue': 2,
'min_lag': 0,
'max_lag': 0,
Expand All @@ -256,3 +263,52 @@ def test_suppress_work_in_progress(self):
'epidata': expected,
'message': 'success',
})


def test_min_issue(self):
"""Test proper computation of min issue in a complex setup."""

def add_row(src, sig, time_type, geo_type, time_value, geo_value, value, issue, is_latest = True):

template = '''
insert into covidcast(
`id`, `source`, `signal`, `time_type`, `geo_type`, `time_value`, `geo_value`,
`value_updated_timestamp`, `value`, `stderr`, `sample_size`, `direction_updated_timestamp`, `direction`,
`issue`, `lag`, `is_latest_issue`, `is_wip`)
values
(0, "%s", "%s", "%s", "%s", %d, "%s",
19700101, %d, 0, 0, 19700101, 0,
%d, 0, %d, 0)
'''
self.cur.execute(template % (src, sig, time_type, geo_type, time_value, geo_value, value, issue, is_latest))

expected = []
for src in ('src1', 'src2'):
for sig in ('sig1', 'sig2'):
expected.append(dict(data_source=src, signal=sig, min_issue=20200502, max_issue=20200505, mean_value=15))

for time_value in [20200101, 20200102]:
for geo_value, value in zip(('geo1', 'geo2'), (1, 2)):
# add some old issue rows which won't influence the mean
add_row(src, sig, 'day', 'county', time_value, geo_value, value, 20200502, False)

for time_value in range(20200101, 20200105):
for geo_value, value in zip(('geo1', 'geo2'), (10, 20)):
# add a latest issue row
add_row(src, sig, 'day', 'county', time_value, geo_value, value, 20200505, True)

self.cnx.commit()
update_cache(args=None)

# make the request
response = requests.get(BASE_URL, params=dict(source='covidcast_meta', fields="data_source,signal,min_issue,max_issue,mean_value"))
response.raise_for_status()
response = response.json()

# assert that the right data came back
self.maxDiff = None
self.assertEqual(response, {
'result': 1,
'epidata': expected,
'message': 'success',
})
95 changes: 48 additions & 47 deletions src/acquisition/covidcast/database.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -163,12 +163,12 @@ def insert_or_update_batch(self, cc_rows, batch_size=2**20, commit_partial=False
SET `is_latest_issue`=0
'''
set_is_latest_issue_sql = f'''
UPDATE
UPDATE
(
SELECT `source`, `signal`, `time_type`, `geo_type`, `time_value`, `geo_value`, MAX(`issue`) AS `issue`
FROM
(
SELECT DISTINCT `source`, `signal`, `time_type`, `geo_type`, `time_value`, `geo_value`
SELECT DISTINCT `source`, `signal`, `time_type`, `geo_type`, `time_value`, `geo_value`
FROM `{tmp_table_name}`
) AS TMP
LEFT JOIN `covidcast`
Expand All @@ -177,7 +177,7 @@ def insert_or_update_batch(self, cc_rows, batch_size=2**20, commit_partial=False
) AS TMP
LEFT JOIN `covidcast`
USING (`source`, `signal`, `time_type`, `geo_type`, `time_value`, `geo_value`, `issue`)
SET `is_latest_issue`=1
SET `is_latest_issue`=1
'''


Expand Down Expand Up @@ -268,68 +268,69 @@ def get_covidcast_meta(self):

sql = 'SELECT `source`, `signal` FROM covidcast WHERE NOT `is_wip` GROUP BY `source`, `signal` ORDER BY `source` ASC, `signal` ASC;'
self._cursor.execute(sql)
for source, signal in [ss for ss in self._cursor]: #NOTE: this obfuscation protects the integrity of the cursor; using the cursor as a generator will cause contention w/ subsequent queries
signals = [ss for ss in self._cursor] #NOTE: this obfuscation protects the integrity of the cursor; using the cursor as a generator will cause contention w/ subsequent queries
for source, signal in signals:

# calculate the min issues per combination
sql = '''
SELECT
`time_type`,
`geo_type`,
MIN(`issue`) as `min_issue`
FROM
`covidcast`
WHERE
`source` = %s AND
`signal` = %s
GROUP BY
`time_type`,
`geo_type`
'''
self._cursor.execute(sql, (source, signal))
min_issue_lookup = {(x[0], x[1]): x[2] for x in self._cursor}

# calculate statistics for the latest issue entries
sql = '''
SELECT
t.`source` AS `data_source`,
t.`signal`,
t.`time_type`,
t.`geo_type`,
MIN(t.`time_value`) AS `min_time`,
MAX(t.`time_value`) AS `max_time`,
COUNT(DISTINCT t.`geo_value`) AS `num_locations`,
`source` AS `data_source`,
`signal`,
`time_type`,
`geo_type`,
MIN(`time_value`) AS `min_time`,
MAX(`time_value`) AS `max_time`,
COUNT(DISTINCT `geo_value`) AS `num_locations`,
MIN(`value`) AS `min_value`,
MAX(`value`) AS `max_value`,
ROUND(AVG(`value`),7) AS `mean_value`,
ROUND(STD(`value`),7) AS `stdev_value`,
MAX(`value_updated_timestamp`) AS `last_update`,
MIN(`issue`) as `min_issue`,
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

i believe this is not going to work as intended, and instead will show the issue date of the least-recently-updated of the issues in a group, which i would argue is meaningless (plz correct me if im wrong). i think it will require a MIN(issue) inside the sub-SELECT and appropriate handling outside of that.

also, some non-trivial test cases would be nice, with different issue values.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

true, I will think about it how to fix it. Moreover, we can think of using the is_latest_issue to find the max issue

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@melange396 can you take a look at the new statements. I split it into two, since I couldn't think of a fast way to handle both cases at the same time. Moreover, I switched the nested statement to the is_latest_issue flag. It is not that easy to test for me, since I don't have access to the real database but just local test data

Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

i suppose this will work, but the old SQL query would have done the trick if you added min(issue) min_issue`` to the nested select... i think that would be preferable to adding a bunch of extra lines unless you can think of another benefit that justifies the extra complexity

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yes but I think this version should be more efficient using is_latest_issue instead of a join

Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

after wrestling with and waiting for long-duration queries for the past week, i think it might be wise to test the performance of the one-query vs two-query approaches on the staging server. i can certainly help do this if necessary.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Apologies for being so late to this conversation.

The use case for min_issue is briefly described in the source issue for this PR, but the jist is this: A researcher is constructing as_of queries and wants to know the earliest date they can pass in and still expect a non-nil result. This means that min_issue must not be restricted to rows with is_latest_issue=1, since that would hide issues which are not the most recent but nevertheless accessible using as_of.

Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

i will go ahead and try this and other variants out on the staging copy of the database to check performance and runtimes

MAX(`issue`) as `max_issue`,
MIN(`lag`) as `min_lag`,
MAX(`lag`) as `max_lag`
FROM
`covidcast` t
JOIN
(
SELECT
max(`issue`) `max_issue`,
`time_type`,
`time_value`,
`source`,
`signal`,
`geo_type`,
`geo_value`
FROM
`covidcast`
WHERE
`source` = %s AND
`signal` = %s
GROUP BY
`time_value`,
`time_type`,
`geo_type`,
`geo_value`
) x
ON
x.`max_issue` = t.`issue` AND
x.`time_type` = t.`time_type` AND
x.`time_value` = t.`time_value` AND
x.`source` = t.`source` AND
x.`signal` = t.`signal` AND
x.`geo_type` = t.`geo_type` AND
x.`geo_value` = t.`geo_value`
`covidcast`
WHERE
`source` = %s AND
`signal` = %s AND
`is_latest_issue` is TRUE
GROUP BY
t.`time_type`,
t.`geo_type`
`time_type`,
`geo_type`
ORDER BY
t.`time_type` ASC,
t.`geo_type` ASC
`time_type` ASC,
`geo_type` ASC
'''
self._cursor.execute(sql, (source, signal))
meta.extend(list(dict(zip(self._cursor.column_names,x)) for x in self._cursor))

for x in self._cursor:
entry = dict(zip(self._cursor.column_names, x))
# merge in the min issue
key = (entry['time_type'], entry['geo_type'])
entry['min_issue'] = min_issue_lookup.get(key, entry['max_issue'])
meta.append(entry)
return meta

def update_covidcast_meta_cache(self, metadata):
"""Updates the `covidcast_meta_cache` table."""

Expand Down