feat(query): add history-based data lineage

[youngsofun]

I hereby agree to the terms of the CLA available at: https://docs.databend.com/dev/policies/cla/

Summary

Add history-based object and column lineage for successful Databend queries, and expose bounded upstream/downstream traversal through GET_LINEAGE.

Query execution remains authoritative: lineage extraction, logging, and asynchronous history aggregation are best effort and never change the result of the DDL or DML that produced the data.

Concepts and contract

Every lineage edge has one fixed data-flow orientation:

source -> target

• Source is an object or column whose value contributes data.
• Target is the object or column that receives or defines that data.
• UPSTREAM starts at a target and walks backward toward sources.
• DOWNSTREAM starts at a source and walks forward toward consumers.

Traversal direction never reverses the meaning or output order of source and target.

The SQL interface is:

GET_LINEAGE(
    '<object_name>',
    'TABLE' | 'VIEW' | 'STAGE' | 'COLUMN',
    'UPSTREAM' | 'DOWNSTREAM'
    [, <distance> ]
)

distance defaults to 5 and must be between 1 and 5. COLUMN input uses table.column, with normal catalog/database qualification and quoted identifiers supported.

The result contains:

distance
source_object_domain, source_object_name, source_column_name
target_object_domain, target_object_name, target_column_name
target_status
process

Only active, visible edges are returned. For COLUMN results, target_status is MASKED when the resolved target column currently has a masking policy; otherwise it is ACTIVE. Object-level results remain ACTIVE. The status always describes the target side of the returned source -> target edge, independent of whether the query walks UPSTREAM or DOWNSTREAM. process is JSON containing query_id, query_kind, lineage_kind, and event_time.

Architecture

Resolved/optimized Plan
  -> Plan::query_lineage()
  -> QueryContext
  -> successful query-finish or DDL completion
  -> databend::log::lineage JSON
  -> system_history.log_history batch
  -> replay-safe MERGE
  -> system_history.lineage_unresolved
  -> GET_LINEAGE frontier scans
  -> live object/column resolution and visibility checks
  -> shortest-path deduplication and stable output

Planner extraction

The resolver starts at each target value expression and follows optimizer symbol definitions back to base relation columns. It records only columns that contribute to target values; filter-only, join-only, ordering-only, and WHEN-only columns are excluded unless they also contribute to a written value. count(col) records col, while count(*) has no column source.

The canonical in-memory model is:

QueryLineage {
    kind: Ctas | Dml | CreateView,
    targets: Vec<LineageTarget>,
}

Relations carry catalog, database, name, optional stable ID, catalog type, and relation kind (TABLE, VIEW, or STAGE). Columns carry both name and column ID; persistence selects the stable representation appropriate for the endpoint.

View and Stream semantics are deliberately different:

• A View is a lineage boundary. A query reading a View references the View's output columns rather than directly inheriting the View definition's base-table expressions. The View definition is represented by a separate edge.
• A Stream is transparent. Its data columns resolve to the backing table, and Stream metadata columns are excluded.

Successful-query logging

Extraction runs only when lineage_unresolved history is enabled. Extraction failures are logged as warnings and do not fail the query. Captured lineage is emitted only after successful pipeline completion or successful empty-pipeline DDL completion.

Each event records source/target identities, query/process fields, and two normalized column maps:

source_to_target_columns: MAP(STRING, ARRAY(STRING))
target_to_source_columns: MAP(STRING, ARRAY(STRING))

Column pairs are sorted and deduplicated before a stable SHA-256 is calculated. Stage edges intentionally carry empty column maps because stage file fields are not a stable schema. Self-loops are removed after endpoint addressing is known.

History aggregation

lineage_unresolved is merged by:

source_lineage_key
+ target_lineage_key
+ lineage_kind
+ column_lineage_hash

Repeated events with the same physical pattern update the latest query/process time. Different column mappings remain separate rows. CTAS, CREATE VIEW, and DML remain distinct because lineage_kind participates in the identity.

Lineage has no retention by default. If retention is configured, only old DML patterns expire; CTAS and CREATE VIEW records remain. ID-addressed deletion tombstones asynchronously remove edges for permanently deleted objects.

Addressing rules

Endpoint	Object address	Column address	Expected behavior
Default-catalog physical table	stable table ID	stable column ID	table/column rename remains connected; undrop restores visibility
CREATE VIEW upstream object	qualified name	column name	referenced rename breaks the edge; renaming back restores it
View target	stable table ID	column name	View output ordinals are not treated as stable schema IDs
View consumed by DML	stable table ID	column name	traversal stops at the View boundary
Iceberg/Hive/Paimon	qualified name	column name	runtime IDs are not persisted; endpoint is terminal in v1
Named Stage	stage name	none	object-level lineage only
Stream	not persisted as an endpoint	backing-table addressing	transparent pass-through
MEMORY, DELTA, temporary table	skipped	skipped	no durable lineage identity in v1

Default-catalog keys use forms such as TABLE::ID::42; name-addressed keys use TABLE::NAME::catalog.database.table; Stage keys use STAGE::NAME::stage_name.

GET_LINEAGE execution

GET_LINEAGE is an opaque table function backed by one AsyncSource.

1. Resolve the starting object according to the requested domain and current SQL name-resolution settings.
2. Pin one lineage_unresolved table instance so all levels read one Fuse snapshot.
3. For UPSTREAM, filter target_lineage_key; for DOWNSTREAM, filter source_lineage_key.
4. Batch each frontier into at most 256 keys and push the key predicates into the table scan.
5. Resolve both captured endpoints against current catalog/meta state and object visibility.
6. Resolve only matching column-map entries for COLUMN traversal and derive target_status from the target column's current masking-policy metadata.
7. Deduplicate each generation, reject cycles and self-loops, and advance expandable endpoints.
8. Prefer the shortest distance; for equal semantic edges choose the newest process event; then sort output deterministically.

Default-catalog ID resolution follows TableIdToName -> database ID -> current table and verifies the loaded table ID before accepting it. Name-addressed objects must still resolve with the expected TABLE/VIEW type. Stages must currently exist and be visible.

External catalog endpoints are name-addressed and terminal in v1. They can be returned as one end of an edge, but traversal does not continue through catalog-specific internal identities.

Each frontier batch is read through a child QueryContext, local pipeline, and PipelinePullingExecutor. The pulling stream runs on an isolated two-thread runtime to avoid starving the outer async source. This follows the existing Recursive CTE nested-pipeline precedent, but the scheduler, cancellation, profiling, runtime creation, and batch materialization behavior deserve focused review.

Lifecycle behavior

• Physical table and column rename preserve ID-addressed lineage and display current names.
• Dropping a normal Fuse table hides its edges because the object no longer resolves; undrop restores the same ID and naturally restores the edges.
• Dropping a transient table or View emits an immediate deletion tombstone.
• VACUUM emits tombstones when dropped Fuse object metadata is permanently removed.
• Dropping and adding a same-named column allocates a new ID, so the replacement column does not inherit old lineage.
• CREATE OR REPLACE uses the new target ID; old edges do not attach to the replacement object.
• View dependencies are name-addressed by design, so referenced object or column rename breaks the definition edge until the name returns.
• Name-addressed Stage or external objects may reactivate old edges after same-name recreation; this is a v1 limitation.

Supported statements and expressions

The implementation covers CREATE VIEW, CTAS, INSERT SELECT, multi-table INSERT, REPLACE, UPDATE/MERGE mutation paths, COPY to/from named Stage, INSERT FROM @stage, and INSERT SELECT FROM @stage.

Planner coverage includes aliases, casts, scalar expressions, aggregate arguments, subqueries, UNION ALL, ordinary CTEs, automatically materialized CTEs, View boundaries, Stream pass-through, and self-insert suppression.

Column lineage also reports whether each resolved target column is currently protected by a masking policy. Masking is a live status annotation: it does not hide the edge, change traversal, or mark the containing table as masked.

Known limitations and review focus

• History ingestion is asynchronous and best effort; successful SQL may temporarily or permanently lack lineage if logging/ingestion fails.
• Existing objects and historical DML are not backfilled when lineage is enabled.
• Explicit AS MATERIALIZED CTE producer-column expansion is not supported.
• Stage column lineage, external-catalog internal multi-hop traversal, and paths beyond five hops are not supported.
• The lineage event stores query ID and basic process fields, but not query text, task, procedure, or full process context.
• A normal INSERT still re-resolves its execution target by name, but its captured lineage target ID originates at bind time. A DROP/REPLACE race between planning and execution needs particular correctness review.
• Access to system_history requires the SystemHistory enterprise feature. Configuring lineage_unresolved as invisible currently also prevents GET_LINEAGE from reading it.
• The raw history table is an implementation surface and can contain stale edges; consumers should not treat every physical row as active without applying the same resolution rules.
• The pipeline implementation materializes each frontier scan's blocks before advancing and creates an isolated runtime per scan batch. Resource bounds and cancellation should be reviewed carefully.

Tests

• Unit Test
• Logic Test
• Benchmark Test
• No Test - Explain why

Unit coverage includes planner extraction, View/Stream semantics, CTEs, aggregates, subqueries, Stage sources, catalog/engine addressing, normalized bidirectional column maps, hash identity, self-loop filtering, history MERGE replay behavior, tombstones, retention, argument parsing, target masking status, and traversal helpers.

History sqllogic coverage includes object and column traversal in both directions, multi-hop and multiple mapping patterns, column-level MASKED status without changing object-level ACTIVE status, View boundaries, Stream pass-through, CTAS/DML/COPY/multi-insert/replace, Stage endpoints, self-insert, rename/drop/undrop/replace/vacuum behavior, column replacement, and an Iceberg REST catalog fixture. History readiness is polled rather than relying on one fixed sleep.

Static validation before publishing this PR:

• cargo fmt --all -- --check
• git diff --check up/main...HEAD
• bash -n for the modified history test scripts

Type of change

• Bug Fix (non-breaking change which fixes an issue)
• New Feature (non-breaking change which adds functionality)
• Breaking Change (fix or feature that could cause existing functionality not to work as expected)
• Documentation Update
• Refactoring
• Performance Improvement
• Other (please describe):

AI assistance

• AI usage: An AI coding agent assisted with implementation, test coverage, local validation, commit organization, and drafting the review-oriented design summary.
• Responsible human: @youngsofun
• The responsible human has read every line of this diff and can explain each change

---

This change is