mobuchowski commented on code in PR #74805:
URL: https://github.com/apache/airflow/pull/74805#discussion_r4375460605
##########
providers/openlineage/src/airflow/providers/openlineage/utils/sql_hook_lineage.py:
##########
@@ -133,7 +133,8 @@ def _get_database_info(hook_id: int):
try:
return hook.get_openlineage_database_info(_get_connection(hook_id))
Review Comment:
`_get_database_info` still calls `hook.get_openlineage_database_info(...)`
directly inside `except Exception`. The other two call sites in this PR now use
`getattr(..., None)`. Since this branch now logs at WARNING, a hook that simply
doesn't implement the method also produces a warning.
Several hooks send SQL hook lineage without being `DbApiHook` subclasses and
without defining the method: `AthenaHook` (`athena.py:138`), `RedshiftDataHook`
(`redshift_data.py:158`) and `HiveCliHook` (`hive.py:372`). `_get_hook_conn_id`
resolves a conn id for them through `conn_name_attr` (e.g. `aws_default`), so
`_get_database_info` runs. It hits `AttributeError: 'AthenaHook' object has no
attribute 'get_openlineage_database_info'` and now logs:
`WARNING Failed to get OpenLineage database info for aws_default:
'AthenaHook' object has no attribute ...`
That shows up on every task that uses these hooks. It's a supported "no
database info" case, not a failure, and it looks exactly like the real errors
this PR is trying to surface.
Suggest using the same lookup as the other call sites:
```python
get_database_info = getattr(hook, "get_openlineage_database_info", None)
if get_database_info is None:
return None
try:
return get_database_info(_get_connection(hook_id))
except Exception as e:
log.warning(...)
```
##########
providers/openlineage/tests/unit/openlineage/utils/test_sql_hook_lineage.py:
##########
@@ -191,6 +191,42 @@ def
test_sql_parsing_failure_falls_back_to_normalized_sql_facet(self):
assert call.kwargs["job_facets"]["sql"].query ==
SQLParser.normalize_sql("SELECT broken(")
assert call.kwargs["run_facets"]["externalQuery"].externalQueryId ==
"qid-1"
+ def test_sql_parsing_failure_is_logged_as_warning(self, caplog):
+ self.mock_ns.return_value = "ns"
+ self.mock_facets_fn.side_effect = Exception("parse error")
+ mock_ti = mock.MagicMock(dag_id="dag_id", task_id="task_id")
+
+ with caplog.at_level(logging.WARNING, logger=_MODULE):
+ emit_lineage_from_sql_extras(
+ task_instance=mock_ti,
+ sql_extras=[_make_extra(sql="SELECT broken(", job_id="qid-1")],
+ )
+
+ assert "Failed to parse SQL for query 1: parse error" in caplog.text
Review Comment:
The repo guidelines ask not to assert on `caplog.text`, since those
string-matching APIs are slated for removal. Structured checks are preferred
(`"event" in caplog` / `{"event": ..., ...} in caplog`). The same pattern
appears at L219 here and at `test_manager.py:387-388`. Optional, since the
existing tests in this file already do it.
##########
providers/openlineage/src/airflow/providers/openlineage/sqlparser.py:
##########
@@ -489,10 +489,8 @@ def get_openlineage_facets_with_sql(
connection = hook.get_connection(conn_id)
if database_info is None:
- try:
- database_info = hook.get_openlineage_database_info(connection)
- except AttributeError:
- database_info = None
+ get_database_info = getattr(hook, "get_openlineage_database_info",
None)
+ database_info = get_database_info(connection) if get_database_info is
not None else None
Review Comment:
Nothing I know of triggers this today, but the same bug class this PR fixes
remains a few lines below and in the operator. `DbApiHook` defines
`get_openlineage_database_dialect`, `get_openlineage_default_schema` and
`get_openlineage_database_specific_lineage` with defaults, so for SQL hooks
these `except AttributeError` blocks never mean "method missing". They only
ever catch an `AttributeError` raised *inside* the method, log it at debug and
return no lineage:
- `sqlparser.py:499-505`: dialect and default schema, logged as "failed to
get database dialect"
- `operators/sql.py:307-313`: the same pair in `SQLExecuteQueryOperator`
- `operators/sql.py:337-340`: `get_openlineage_database_specific_lineage`,
logged as "has no database specific lineage provided"
For example, `SnowflakeHook.get_openlineage_default_schema` calls
`_get_conn_params()`. A broken import there would silently produce empty
lineage again. Worth switching these to the same `getattr` pattern in this PR
for consistency? Optional.
##########
providers/common/sql/tests/unit/common/sql/operators/test_sql_execute.py:
##########
@@ -368,6 +368,40 @@ def get_db_hook(self):
)
+def test_openlineage_hook_without_database_info_method_returns_empty_lineage():
+ from airflow.providers.openlineage.extractors import OperatorLineage
+
+ hook = mock.MagicMock(spec=["get_connection", "conn_name_attr", "conn_id"])
+ hook.conn_name_attr = "conn_id"
+ hook.conn_id = "sql_default"
+
+ class SQLExecuteQueryOperatorForTest(SQLExecuteQueryOperator):
+ def get_db_hook(self):
+ return hook
+
+ op = SQLExecuteQueryOperatorForTest(task_id=TASK_ID, sql="SELECT 1;")
+
+ assert op.get_openlineage_facets_on_start() == OperatorLineage()
+
+
+def test_openlineage_attribute_error_from_database_info_is_not_swallowed():
+ hook = mock.MagicMock()
Review Comment:
Small one: the guidelines ask for `spec`/`autospec` on mocks.
`mock.MagicMock(spec=DbApiHook)` would work here, in
`test_sqlparser.py::test_attribute_error_from_database_info_is_not_swallowed`
and in `test_sql_hook_lineage.py:208`. It also means a typo in the method name
can't make the test pass by accident.
##########
providers/common/sql/src/airflow/providers/common/sql/operators/sql.py:
##########
@@ -297,11 +297,8 @@ def get_openlineage_facets_on_start(self) ->
OperatorLineage | None:
use_external_connection = True
connection = hook.get_connection(getattr(hook, hook.conn_name_attr))
- try:
- database_info = hook.get_openlineage_database_info(connection)
- except AttributeError:
- self.log.debug("%s has no database info provided", hook)
- database_info = None
+ get_database_info = getattr(hook, "get_openlineage_database_info",
None)
+ database_info = get_database_info(connection) if get_database_info is
not None else None
Review Comment:
Before this PR, an `AttributeError` inside `get_openlineage_database_info`
turned into `database_info = None`, and the callers kept whatever lineage they
could build without it. Now the exception leaves the operator's facets method
entirely. `DefaultExtractor._get_openlineage_facets` catches it and returns
`None`, so **all** operator lineage for that task is lost, not just the
SQL-parsing part:
- `SQLExecuteQueryOperator.get_openlineage_facets_on_complete` calls
`get_openlineage_facets_on_start()` (L335) without a guard. It no longer
reaches `get_openlineage_database_specific_lineage` (query IDs, external query
facets) or `_attach_check_facets` for check operators. The new test asserts
this (`pytest.raises` on `on_complete`).
- `MySQLToGCSOperator`, `PostgresToGCSOperator`, `OracleToGCSOperator`,
`MSSQLToGCSOperator`, `TrinoToGCSOperator`, `CloudSQLExecuteQueryOperator` and
`SpannerQueryDatabaseInstanceOperator` all call
`get_openlineage_facets_with_sql`. Some of them, like
`mysql_to_gcs.py:155-170`, have a deliberate fallback: when SQL parsing returns
`None`, they still emit the GCS output datasets and the SQL facet. Now that
fallback never runs, so those tasks lose their GCS outputs too.
Other exception types already propagated this way, so this makes
`AttributeError` consistent with them. Is losing the non-SQL lineage the
intended tradeoff? The goal is a visible warning. You could get that while
keeping partial lineage by catching `Exception` at these two call sites,
logging at WARNING and falling back to `None`, which is what
`sql_hook_lineage._get_database_info` already does.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
