Move lineage from airflow core to task sdk

[amoghrajesh]

---

Was generative AI tooling used to co-author this PR?

• No

---

Why?

Lineage collection is a task execution concern and on checking it only runs on workers (task sdk consumer) processes, not in any server components (scheduler, api server). I intend to move lineage module from airflow-core to task-sdk as part of the ongoing client server separation work.

Some more context:

• io/path.py intercepts file I/O during task execution
• OpenLineage listeners run in the worker process after task completion
• It only orchestrates, never executes user code

What is done?

• Created a new module in sdk: sdk/lineage.py
• Moved all classes from airflow.lineage.hook
  • Updated imports to SDK equivalents:
    • ProvidersManager → ProvidersManagerTaskRuntime
    • airflow.utils.log.logging_mixin → airflow.sdk.definitions._internal.logging_mixin
  • Created get_hook_lineage_readers_plugins() using SDK's plugin discovery

Backward Compatibility

1. For core -

   • Supports both from airflow.lineage.hook import X and from airflow.lineage import hook

2. Provider compatibility has been handled with providers/common/compat/src/airflow/providers/common/compat/sdk.py

   • Added lineage classes to compat layer
   • Handles DatasetLineageInfo → AssetLineageInfo rename (AF2 → AF3)

3. Removed from core -airflow-core/src/airflow/plugins_manager.py

   • Deleted unused get_hook_lineage_readers_plugins() function
   • Function only existed for core's old lineage implementation

4. For provider developers, it is recommended to use imports from airflow.providers.common.compat.sdk

Testing

To gain confidence I tried to test a manual e2e scenario for this. Ran breeze with OL integration:
breeze start-airflow --integration openlineage

DAG:

from __future__ import annotations

from datetime import datetime

from airflow.sdk import DAG
from airflow.hooks.base import BaseHook
from airflow.providers.standard.operators.python import PythonOperator


from airflow.sdk.lineage import get_hook_lineage_collector


class SimpleHook(BaseHook):

    def process(self):
        collector = get_hook_lineage_collector()

        collector.add_input_asset(self, uri="file:///input/data.csv")

        collector.add_output_asset(self, uri="file:///output/result.csv")

        print("Lineage reported! "*10)


def my_task():
    hook = SimpleHook()
    hook.process()


with DAG(
    dag_id="simple_lineage",
    start_date=datetime(2021, 1, 1),
    schedule=None,
    catchup=False,
) as dag:
    PythonOperator(
        task_id="run_hook",
        python_callable=my_task,
    )

Its a simple dag that does this:

• Creates a custom hook (SimpleHook) that reports lineage information
• Reports input dataset: file:///input/data.csv
• Reports output dataset: file:///output/result.csv
• Sends lineage to OpenLineage by using get_hook_lineage_collector() to register assets

DAG run:

Marquez:

[mobuchowski]

That looks good to me, thanks @amoghrajesh!
Couple of tests to fix though.

[potiuk]

One thing about # use next version

[jason810496]

Nice! LGTM once the CI pass.

[amoghrajesh]

CI's been a challenge on this one, getting there

[github-actions]

Backport failed to create: v3-1-test. View the failure log Run details

Status	Branch	Result
❌	v3-1-test

You can attempt to backport this manually by running:

cherry_picker e871af6 v3-1-test

This should apply the commit to the v3-1-test branch and leave the commit in conflict state marking
the files that need manual conflict resolution.

After you have resolved the conflicts, you can continue the backport process by running:

cherry_picker --continue

If you don't have cherry-picker installed, see the installation guide.

[amoghrajesh]

No need to backport.

[jscheffl]

@amoghrajesh I am terribly sorry but canary failed and needed to revert. Can you re-apply and make "all versions" and "full tests" as labels? Seems the test selection did not catch all.