Skip to content

Add openlineage system tests for missing use cases #68664

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

Merged
mobuchowski merged 1 commit into from
Jun 17, 2026
Merged

Add openlineage system tests for missing use cases #68664

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
2 changes: 2 additions & 0 deletions docs/spelling_wordlist.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1148,6 +1148,8 @@ ok
oklch
Okta
okta
OL
ol
Ollama
onboarded
onboarding
Expand Down
2 changes: 1 addition & 1 deletion generated/provider_dependencies.json.sha256sum
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1 +1 @@
8609061b1d7c65722ca143c6e54bf569c2b3bb2bfeac9ecc85c97a114a5d83ac
bb7437125421517dcc83ca840e1c068e25179eff8aab93b87766ec29d0dfa3b0
11 changes: 6 additions & 5 deletions providers/openlineage/src/airflow/providers/openlineage/api/datasets.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,7 @@

from airflow.models.taskinstance import TaskInstance
from airflow.sdk.execution_time.task_runner import RuntimeTaskInstance
from airflow.sdk.types import RuntimeTaskInstanceProtocol

log = logging.getLogger(__name__)

Expand All @@ -56,9 +57,9 @@

def emit_dataset_lineage(
*,
inputs: list[InputDataset] | None = None,
outputs: list[OutputDataset] | None = None,
task_instance: RuntimeTaskInstance | TaskInstance | None = None,
inputs: list[InputDataset | Dataset] | None = None,
outputs: list[OutputDataset | Dataset] | None = None,
task_instance: RuntimeTaskInstanceProtocol | RuntimeTaskInstance | TaskInstance | None = None,
additional_run_facets: dict[str, RunFacet] | None = None,
additional_job_facets: dict[str, JobFacet] | None = None,
raise_on_error: bool = False,
Expand Down Expand Up @@ -159,8 +160,8 @@ def my_task():
name=lineage_job_name(task_instance),
facets=job_facets,
),
inputs=inputs,
outputs=outputs,
inputs=inputs, # type: ignore[arg-type]
outputs=outputs, # type: ignore[arg-type]
producer=_PRODUCER,
)

Expand Down
9 changes: 5 additions & 4 deletions providers/openlineage/src/airflow/providers/openlineage/api/sql.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,11 +38,12 @@
if TYPE_CHECKING:
from datetime import datetime

from openlineage.client.event_v2 import Dataset
from openlineage.client.event_v2 import Dataset, InputDataset, OutputDataset
from openlineage.client.facet_v2 import JobFacet, RunFacet

from airflow.models.taskinstance import TaskInstance
from airflow.sdk.execution_time.task_runner import RuntimeTaskInstance
from airflow.sdk.types import RuntimeTaskInstanceProtocol

log = logging.getLogger(__name__)

Expand All @@ -54,16 +55,16 @@ def emit_query_lineage(
query_id: str | None = None,
query_source_namespace: str | None = None,
query_text: str | None = None,
inputs: list[Dataset] | None = None,
outputs: list[Dataset] | None = None,
inputs: list[InputDataset | Dataset] | None = None,
outputs: list[OutputDataset | Dataset] | None = None,
start_time: datetime | None = None,
end_time: datetime | None = None,
is_successful: bool = True,
error_message: str | None = None,
default_database: str | None = None,
default_schema: str | None = None,
job_name: str | None = None,
task_instance: TaskInstance | RuntimeTaskInstance | None = None,
task_instance: RuntimeTaskInstanceProtocol | RuntimeTaskInstance | TaskInstance | None = None,
additional_run_facets: dict[str, RunFacet] | None = None,
additional_job_facets: dict[str, JobFacet] | None = None,
raise_on_error: bool = False,
Expand Down
Loading
Loading