Skip to content

Convert covidcast signal OR clauses to UNION #1021

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

Closed
wants to merge 5 commits into from
Closed

Convert covidcast signal OR clauses to UNION #1021

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
e348aab
1. Refactored the QueryBuilder code to convert covidcast signal OR cl…
LeonLu2 Nov 2, 2022
e77d7c4
Merge branch 'dev' into leonlu2/or-to-unionall
LeonLu2 Nov 2, 2022
978e009
1. changed some formatting and cleaned up the code
LeonLu2 Nov 5, 2022
a853149
Merge branch 'dev' into leonlu2/or-to-unionall
LeonLu2 Nov 5, 2022
5548d3c
Merge branch 'dev' into leonlu2/or-to-unionall
LeonLu2 Nov 8, 2022
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
78 changes: 66 additions & 12 deletions src/server/_query.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,8 +10,8 @@
Tuple,
Union,
cast,
Mapping,
)
Mapping, Type,
)

from sqlalchemy import text
from sqlalchemy.engine import Row
Expand Down Expand Up @@ -65,6 +65,20 @@ def filter_values(
conditions = [to_condition(field, v, f"{param_key}_{i}", params, formatter) for i, v in enumerate(values)]
return f"({' OR '.join(conditions)})"

def alternative_filter_values(
field: str,
values: Optional[Sequence[Union[Tuple[str, str], str, Tuple[int, int], int]]],
param_key: str,
params: Dict[str, Any],
formatter=lambda x: x,
):
if not values:
return "FALSE"
# builds a SQL expression to filter strings (ex: locations)
# $field: name of the field to filter
# $values: array of values
conditions = [to_condition(field, v, f"{param_key}_{i}", params, formatter) for i, v in enumerate(values)]
return conditions

def filter_strings(
field: str,
Expand All @@ -74,6 +88,14 @@ def filter_strings(
):
return filter_values(field, values, param_key, params)

def alternative_filter_strings(
field: str,
values: Optional[Sequence[Union[Tuple[str, str], str]]],
param_key: str,
params: Dict[str, Any],
):
return alternative_filter_values(field, values, param_key, params)


def filter_integers(
field: str,
Expand Down Expand Up @@ -150,25 +172,37 @@ def filter_source_signal_pairs(
values: Sequence[SourceSignalPair],
param_key: str,
params: Dict[str, Any],
) -> str:
) -> Union[str, list]:
"""
returns the SQL sub query to filter by the given source signal pairs
returns the SQL sub queries in an array to filter by the given source signal pairs
"""

def filter_pair(pair: SourceSignalPair, i) -> str:
def filter_pair(pair: SourceSignalPair, i) -> Union[str, list]:
source_param = f"{param_key}_{i}t"
params[source_param] = pair.source
if isinstance(pair.signal, bool) and pair.signal:
return f"{source_field} = :{source_param}"
return f"({source_field} = :{source_param} AND {filter_strings(signal_field, cast(Sequence[str], pair.signal), source_param, params)})"

parts = [filter_pair(p, i) for i, p in enumerate(values)]
conditions = alternative_filter_strings(signal_field, cast(Sequence[str], pair.signal), source_param, params)
condition_array = []
if conditions:
for condition in conditions:
condition_array.append(f"({source_field} = :{source_param} AND {condition})")
Comment on lines +188 to +190
Copy link
Contributor

Choose a reason for hiding this comment

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

Checking my understanding: this block will always execute, since conditions will always evaluate true: alternative_filter_strings wraps alternative_filter_values, and alternative_filter_values returns either a nonempty string ("False") or a nonempty list (a function of pair.signal)

Is that right?

return condition_array

parts = []
for i, p in enumerate(values):
array = filter_pair(p, i)
if isinstance(array, str):
parts.append(filter_pair(p, i))
else:
for x in array:
parts.append(x)

if not parts:
# something has to be selected
return "FALSE"

return f"({' OR '.join(parts)})"
return parts


def filter_time_pairs(
Expand Down Expand Up @@ -344,6 +378,7 @@ def __init__(self, table: str, alias: str):
self.order: Union[str, List[str]] = ""
self.fields: Union[str, List[str]] = "*"
self.conditions: List[str] = []
self.signal_array: List[str] = []
self.params: Dict[str, Any] = {}
self.subquery: str = ""
self.index: Optional[str] = None
Expand Down Expand Up @@ -375,7 +410,19 @@ def __str__(self):
group_by = f"GROUP BY {_join_l(self.group_by)}" if self.group_by else ""
index = f"USE INDEX ({self.index})" if self.index else ""

return f"SELECT {self.fields_clause} FROM {self.table} {index} {self.subquery} {where} {group_by} {order}"
# if no signal array, assemble the sql and return
if not self.signal_array:
command = f"SELECT {self.fields_clause} FROM {self.table} {index} {self.subquery} {where} {group_by}"
else:
# if there is a signal array, concatenate signals with UNION ALL
command = f"SELECT {self.fields_clause} FROM {self.table} {index} {self.subquery} {where} AND {self.signal_array[0]} {group_by}"
i = 1
while i < len(self.signal_array):
command += f"\nUNION ALL\nSELECT {self.fields_clause} FROM {self.table} {index} {self.subquery} {where} AND {self.signal_array[i]} {group_by}"
i += 1

command = command + f" {order}"
return command

@property
def query(self) -> str:
Expand Down Expand Up @@ -452,7 +499,7 @@ def where_source_signal_pairs(
) -> "QueryBuilder":
fq_type_field = self._fq_field(type_field)
fq_value_field = self._fq_field(value_field)
self.conditions.append(
self.signal_array.extend(
filter_source_signal_pairs(
fq_type_field,
fq_value_field,
Expand Down Expand Up @@ -499,7 +546,14 @@ def to_asc(v: Union[str, bool]) -> str:
return "DESC"
return cast(str, v)

args_order = [f"{self.alias}.{k} ASC" for k in args]
# Use the column name without their table name for the Order By clause since Union All is used
args_order = []
for i in range(len(args)):
# avoid conflict with the reserved word
if args[i] == 'signal':
args_order.append(f"`signal` ASC")
else:
args_order.append(f"{args[i]} ASC")
kw_order = [f"{self.alias}.{k} {to_asc(v)}" for k, v in kwargs.items()]
self.order = args_order + kw_order
return self
Expand Down
10 changes: 5 additions & 5 deletions tests/server/test_query.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -195,21 +195,21 @@ def test_filter_source_signal_pairs(self):
params = {}
self.assertEqual(
filter_source_signal_pairs("t", "v", [SourceSignalPair("src1", True)], "p", params),
"(t = :p_0t)",
["t = :p_0t"]
)
self.assertEqual(params, {"p_0t": "src1"})
with self.subTest("single"):
params = {}
self.assertEqual(
filter_source_signal_pairs("t", "v", [SourceSignalPair("src1", ["sig1"])], "p", params),
"((t = :p_0t AND (v = :p_0t_0)))",
["(t = :p_0t AND v = :p_0t_0)"]
)
self.assertEqual(params, {"p_0t": "src1", "p_0t_0": "sig1"})
with self.subTest("multi"):
params = {}
self.assertEqual(
filter_source_signal_pairs("t", "v", [SourceSignalPair("src1", ["sig1", "sig2"])], "p", params),
"((t = :p_0t AND (v = :p_0t_0 OR v = :p_0t_1)))",
["(t = :p_0t AND v = :p_0t_0)", "(t = :p_0t AND v = :p_0t_1)"]
)
self.assertEqual(params, {"p_0t": "src1", "p_0t_0": "sig1", "p_0t_1": "sig2"})
with self.subTest("multiple pairs"):
Expand All @@ -222,7 +222,7 @@ def test_filter_source_signal_pairs(self):
"p",
params,
),
"(t = :p_0t OR t = :p_1t)",
["t = :p_0t", "t = :p_1t"]
)
self.assertEqual(params, {"p_0t": "src1", "p_1t": "src2"})
with self.subTest("multiple pairs with value"):
Expand All @@ -238,7 +238,7 @@ def test_filter_source_signal_pairs(self):
"p",
params,
),
"((t = :p_0t AND (v = :p_0t_0)) OR (t = :p_1t AND (v = :p_1t_0)))",
["(t = :p_0t AND v = :p_0t_0)", "(t = :p_1t AND v = :p_1t_0)"]
)
self.assertEqual(
params,
Expand Down