Add BigQuery Column and Table Check Operators #26368
Conversation
boring-cyborg
bot
added
area:providers
provider:google
denimalpaca
force-pushed
the
add_bigquery_common_check_operators
branch
from
ba66b90
to
8e76bae
Compare
denimalpaca
force-pushed
the
add_bigquery_common_check_operators
branch
2 times, most recently
from
d31eacc
to
6870aef
Compare
potiuk
force-pushed
the
add_bigquery_common_check_operators
branch
from
6870aef
to
4a5203e
Compare
|
Rebased it to rebuild it @denimalpaca - if you can see if ti works and ping me if it does, that would be awesome |
@potiuk looks like it's all green -- and already behind again |
denimalpaca
force-pushed
the
add_bigquery_common_check_operators
branch
from
4a5203e
to
2113530
Compare
Add two new operators based on the SQLColumnCheckOperator and SQLTableCheckOperator that also provide job_ids so results of the queries can be pulled and parsed, and so OpenLineage can parse datasets and provide lineage information.
denimalpaca
force-pushed
the
add_bigquery_common_check_operators
branch
from
2113530
to
90f217d
Compare
| failed_tests = [] | ||
| for column in self.column_mapping: | ||
| checks = [*self.column_mapping[column]] | ||
| checks_sql = ",".join([self.column_checks[check].replace("column", column) for check in checks]) |
There was a problem hiding this comment.
I know that "self.column_checks" is part of the parent class but I see from git blame that you're the author. So I would recommend replacing the following strings:
"SUM(CASE WHEN column IS NULL THEN 1 ELSE 0 END) AS column_null_check"
with a format string template:
"SUM(CASE WHEN {column} IS NULL THEN 1 ELSE 0 END) AS {column}_null_check"
it can be later used with check.format(column=column) instead of replacing the sub string.
| hook = self.get_db_hook() | ||
| checks_sql = " UNION ALL ".join( | ||
| [ | ||
| self.sql_check_template.replace("check_statement", value["check_statement"]) |
There was a problem hiding this comment.
It will not work with your test data:
checks={"row_count_check": {"check_statement": {"COUNT(*) = 4"}}},
check_statement is set. Did you mean:
checks={"row_count_check": {"check_statement": "COUNT(*) = 4"}}
There was a problem hiding this comment.
Ah, that was a typo on my part in the test case. Yes, it should be the latter.
|
Hi, thank you for the contribution. I noticed it when our system tests started to fail - I left a comment where the possible issue is. But I also see that there are no unit test for those two new operators.. . There could be other issues as well that could be detected by unit tests. Would it be possible for you to add them? @denimalpaca @potiuk Is it possible to revert this change? This code doesn't have tests and may contain other issues. |
| dag_id=self.dag_id, | ||
| task_id=self.task_id, | ||
| logical_date=context["logical_date"], | ||
| configuration=self.configuration, |
There was a problem hiding this comment.
the self.configuration does not exist (or I can't see it). But I see that bigquery allows to pass job_id as empty string. In that case (check other operators) the following method will be invoked to set job_id: _custom_job_id . So you probably don't need to use generate_job_id .
| records.columns = records.columns.str.lower() | ||
| self.log.info("Record:\n%s", records) | ||
|
|
||
| for row in records.iterrows(): |
There was a problem hiding this comment.
I don't know what type record is, but you could try to unpack it for more clarity instead of row[1]:
for _, columns in records.iterrows():
check = columns.get("check_name")
...
|
@bhirsz I've opened a new PR to address your concerns, and we can add some tests in there as well. |
Add two new operators based on the SQLColumnCheckOperator and SQLTableCheckOperator that also provide
job_ids so results of the queries can be pulled and parsed, and so OpenLineage can parse datasets and provide lineage information.closes: #26367