feat(query): persist captured lineage

youngsofun · youngsofun · commit b25ff06b2231 · 2026-07-26T23:03:05.000+08:00
diff --git a/src/query/service/src/interpreters/common/lineage_writer.rs b/src/query/service/src/interpreters/common/lineage_writer.rs
diff --git a/src/query/service/src/interpreters/common/mod.rs b/src/query/service/src/interpreters/common/mod.rs
@@ -15,6 +15,7 @@
 mod column;
 mod finish_hook;
 mod grant;
+mod lineage_writer;
 mod metrics;
 mod notification;
 mod query_log;
@@ -29,6 +30,9 @@ pub mod table_option_validation;
 pub use column::*;
 pub use finish_hook::QueryFinishHooks;
 pub use grant::validate_grant_object_exists;
+pub use lineage_writer::attach_query_lineage_on_finished;
+pub use lineage_writer::build_query_lineage_updates;
+pub use lineage_writer::build_query_lineage_updates_with_lineage;
 pub use log::*;
 pub use notification::get_notification_client_config;
 pub use query_log::InterpreterQueryLog;
diff --git a/src/query/service/src/interpreters/interpreter_copy_into_table.rs b/src/query/service/src/interpreters/interpreter_copy_into_table.rs
@@ -67,6 +67,8 @@ use log::info;
 use crate::interpreters::HookOperator;
 use crate::interpreters::Interpreter;
 use crate::interpreters::SelectInterpreter;
+use crate::interpreters::common::attach_query_lineage_on_finished;
+use crate::interpreters::common::build_query_lineage_updates;
 use crate::interpreters::common::check_deduplicate_label;
 use crate::interpreters::common::dml_build_update_stream_req;
 use crate::physical_plans::CopyIntoTable;
@@ -874,6 +876,7 @@ impl CopyIntoTableInterpreter {
                 path_prefix,
             )?;
 
+            let lineage_updates = build_query_lineage_updates(&ctx, None).await?;
             to_table.commit_insertion(
                 ctx.clone(),
                 main_pipeline,
@@ -884,6 +887,7 @@ impl CopyIntoTableInterpreter {
                 deduplicated_label,
                 table_meta_timestamps,
             )?;
+            attach_query_lineage_on_finished(main_pipeline, lineage_updates);
         }
 
         // Purge files.
diff --git a/src/query/service/src/interpreters/interpreter_factory.rs b/src/query/service/src/interpreters/interpreter_factory.rs
@@ -16,6 +16,7 @@ use std::sync::Arc;
 
 use databend_common_ast::ast::ExplainKind;
 use databend_common_catalog::lock::LockTableOption;
+use databend_common_config::GlobalConfig;
 use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
 use databend_common_sql::binder::ExplainConfig;
@@ -153,6 +154,7 @@ impl InterpreterFactory {
                 error!("Access.denied(v2): {:?}", e);
             }
         })?;
+        initialize_query_lineage(&ctx, plan);
         let mut access_logger = AccessLogger::create(ctx.clone());
         access_logger.log(plan);
         access_logger.output();
@@ -937,3 +939,24 @@ impl InterpreterFactory {
         }
     }
 }
+
+fn initialize_query_lineage(ctx: &QueryContext, plan: &Plan) {
+    ctx.init_query_lineage(|| {
+        if !GlobalConfig::instance()
+            .query
+            .common
+            .lineage
+            .capture_enabled
+        {
+            return None;
+        }
+
+        match plan.query_lineage() {
+            Ok(lineage) => lineage,
+            Err(error) => {
+                log::warn!("failed to extract query lineage: {error:?}");
+                None
+            }
+        }
+    });
+}
diff --git a/src/query/service/src/interpreters/interpreter_insert.rs b/src/query/service/src/interpreters/interpreter_insert.rs
@@ -61,6 +61,8 @@ use crate::clusters::ClusterHelper;
 use crate::interpreters::HookOperator;
 use crate::interpreters::Interpreter;
 use crate::interpreters::InterpreterPtr;
+use crate::interpreters::common::attach_query_lineage_on_finished;
+use crate::interpreters::common::build_query_lineage_updates;
 use crate::interpreters::common::check_deduplicate_label;
 use crate::interpreters::common::dml_build_update_stream_req;
 use crate::physical_plans::ConstantTableScan;
@@ -468,6 +470,8 @@ impl Interpreter for InsertInterpreter {
                     build_query_pipeline_without_render_result_set(&self.ctx, &insert_select_plan)
                         .await?;
 
+                let lineage_updates =
+                    build_query_lineage_updates(&self.ctx, Some(table.get_table_info())).await?;
                 table.commit_insertion(
                     self.ctx.clone(),
                     &mut build_res.main_pipeline,
@@ -478,6 +482,7 @@ impl Interpreter for InsertInterpreter {
                     unsafe { self.ctx.get_settings().get_deduplicate_label()? },
                     table_meta_timestamps,
                 )?;
+                attach_query_lineage_on_finished(&mut build_res.main_pipeline, lineage_updates);
 
                 //  Execute the hook operator.
                 if self.plan.branch.is_none() {
diff --git a/src/query/service/src/interpreters/interpreter_insert_multi_table.rs b/src/query/service/src/interpreters/interpreter_insert_multi_table.rs
@@ -44,6 +44,8 @@ use databend_common_storages_fuse::FuseTable;
 use super::HookOperator;
 use crate::interpreters::Interpreter;
 use crate::interpreters::InterpreterPtr;
+use crate::interpreters::common::attach_query_lineage_on_finished;
+use crate::interpreters::common::build_query_lineage_updates;
 use crate::interpreters::common::dml_build_update_stream_req;
 use crate::physical_plans::CastSchema;
 use crate::physical_plans::ChunkAppendData;
@@ -107,6 +109,9 @@ impl Interpreter for InsertMultiTableInterpreter {
         let physical_plan = self.build_physical_plan(false).await?;
         let mut build_res =
             build_query_pipeline_without_render_result_set(&self.ctx, &physical_plan).await?;
+        let lineage_updates = build_query_lineage_updates(&self.ctx, None).await?;
+        attach_query_lineage_on_finished(&mut build_res.main_pipeline, lineage_updates);
+
         // Execute hook.
         if self
             .ctx
diff --git a/src/query/service/src/interpreters/interpreter_mutation.rs b/src/query/service/src/interpreters/interpreter_mutation.rs
@@ -178,6 +178,7 @@ impl MutationInterpreter {
         let mut builder =
             PhysicalPlanBuilder::new(mutation.metadata.clone(), self.ctx.clone(), dry_run);
         builder.set_mutation_build_info(mutation_build_info);
+        builder.set_query_lineage(self.ctx.get_query_lineage());
         builder
             .build(&self.s_expr, *mutation.required_columns.clone())
             .await
diff --git a/src/query/service/src/interpreters/interpreter_replace.rs b/src/query/service/src/interpreters/interpreter_replace.rs
@@ -47,6 +47,7 @@ use crate::interpreters::HookOperator;
 use crate::interpreters::Interpreter;
 use crate::interpreters::InterpreterPtr;
 use crate::interpreters::SelectInterpreter;
+use crate::interpreters::common::build_query_lineage_updates;
 use crate::interpreters::common::check_deduplicate_label;
 use crate::interpreters::common::dml_build_update_stream_req;
 #[cfg(feature = "storage-stage")]
@@ -358,6 +359,7 @@ impl ReplaceInterpreter {
             });
         }
 
+        let lineage_updates = build_query_lineage_updates(&self.ctx, None).await?;
         root = PhysicalPlan::new(CommitSink {
             input: root,
             snapshot: base_snapshot,
@@ -370,6 +372,7 @@ impl ReplaceInterpreter {
             deduplicated_label: unsafe { self.ctx.get_settings().get_deduplicate_label()? },
             table_meta_timestamps,
             recluster_info: None,
+            lineage_updates,
             meta: PhysicalPlanMeta::new("CommitSink"),
         });
 
diff --git a/src/query/service/src/interpreters/interpreter_table_create.rs b/src/query/service/src/interpreters/interpreter_table_create.rs
@@ -69,6 +69,7 @@ use log::info;
 
 use crate::interpreters::InsertInterpreter;
 use crate::interpreters::Interpreter;
+use crate::interpreters::common::build_query_lineage_updates;
 use crate::interpreters::common::table_option_validation::is_valid_analyze_count_min_sketch_error_rate;
 use crate::interpreters::common::table_option_validation::is_valid_analyze_frequency_columns;
 use crate::interpreters::common::table_option_validation::is_valid_analyze_histogram_algorithm;
@@ -240,6 +241,7 @@ impl CreateTableInterpreter {
             TableIdent::new(table_id, table_id_seq),
             table_meta,
         );
+        let lineage_updates = build_query_lineage_updates(&self.ctx, Some(&table_info)).await?;
 
         let insert_plan = Insert {
             catalog: self.plan.catalog.clone(),
@@ -288,6 +290,7 @@ impl CreateTableInterpreter {
                         "create_table_as_select {} success, commit table meta data by table id {}",
                         qualified_table_name, table_id
                     );
+                    let lineage_updates = lineage_updates.clone();
                     let fut = async move {
                         let req = CommitTableMetaReq {
                             name_ident: TableNameIdent {
@@ -299,6 +302,7 @@ impl CreateTableInterpreter {
                             table_id,
                             prev_table_id,
                             orphan_table_name,
+                            lineage_updates,
                         };
                         catalog.commit_table_meta(req).await
                     };
@@ -536,6 +540,7 @@ impl CreateTableInterpreter {
                 table_name: self.plan.table.to_string(),
             },
             table_meta,
+            lineage_updates: vec![],
             as_dropped: false,
             table_properties: self.plan.table_properties.clone(),
             table_partition: self.plan.table_partition.as_ref().map(|table_partition| {
diff --git a/src/query/service/src/interpreters/interpreter_table_recluster.rs b/src/query/service/src/interpreters/interpreter_table_recluster.rs
@@ -440,6 +440,7 @@ impl ReclusterTableInterpreter {
             deduplicated_label: None,
             table_meta_timestamps,
             recluster_info,
+            lineage_updates: vec![],
             meta: PhysicalPlanMeta::new("CommitSink"),
         })
     }
diff --git a/src/query/service/src/interpreters/interpreter_view_create.rs b/src/query/service/src/interpreters/interpreter_view_create.rs
@@ -27,6 +27,7 @@ use databend_common_storages_basic::view_table::QUERY;
 use databend_common_storages_basic::view_table::VIEW_ENGINE;
 
 use crate::interpreters::Interpreter;
+use crate::interpreters::common::build_query_lineage_updates;
 use crate::interpreters::util::check_view_circular_dependency;
 use crate::pipelines::PipelineBuildResult;
 use crate::sessions::QueryContext;
@@ -115,6 +116,7 @@ impl Interpreter for CreateViewInterpreter {
         };
         options.insert(QUERY.to_string(), subquery);
 
+        let lineage_updates = build_query_lineage_updates(&self.ctx, None).await?;
         let plan = CreateTableReq {
             create_option: self.plan.create_option,
             catalog_name: if self.plan.create_option.is_overriding() {
@@ -132,6 +134,7 @@ impl Interpreter for CreateViewInterpreter {
                 options,
                 ..Default::default()
             },
+            lineage_updates,
             as_dropped: false,
             table_properties: None,
             table_partition: None,
diff --git a/src/query/service/src/physical_plans/physical_commit_sink.rs b/src/query/service/src/physical_plans/physical_commit_sink.rs
@@ -18,6 +18,7 @@ use std::sync::Arc;
 use databend_common_catalog::plan::ReclusterInfoSideCar;
 use databend_common_exception::Result;
 use databend_common_expression::DataSchemaRef;
+use databend_common_meta_app::schema::LineageUpdate;
 use databend_common_meta_app::schema::TableInfo;
 use databend_common_meta_app::schema::UpdateStreamMetaReq;
 use databend_common_pipeline::core::ExecutionInfo;
@@ -35,6 +36,7 @@ use databend_storages_common_table_meta::meta::TableMetaTimestamps;
 use databend_storages_common_table_meta::meta::TableSnapshot;
 use databend_storages_common_table_meta::readers::snapshot_reader::TableSnapshotAccessor;
 
+use crate::interpreters::common::attach_query_lineage_on_finished;
 use crate::physical_plans::physical_plan::IPhysicalPlan;
 use crate::physical_plans::physical_plan::PhysicalPlan;
 use crate::physical_plans::physical_plan::PhysicalPlanMeta;
@@ -52,6 +54,8 @@ pub struct CommitSink {
     pub commit_type: CommitType,
     pub update_stream_meta: Vec<UpdateStreamMetaReq>,
     pub deduplicated_label: Option<String>,
+    // Best-effort lineage updates are attached after the table commit sink is built.
+    pub lineage_updates: Vec<LineageUpdate>,
     pub table_meta_timestamps: TableMetaTimestamps,
 
     // Used for recluster.
@@ -95,6 +99,7 @@ impl IPhysicalPlan for CommitSink {
             commit_type: self.commit_type.clone(),
             update_stream_meta: self.update_stream_meta.clone(),
             deduplicated_label: self.deduplicated_label.clone(),
+            lineage_updates: self.lineage_updates.clone(),
             table_meta_timestamps: self.table_meta_timestamps,
             recluster_info: self.recluster_info.clone(),
         })
@@ -211,7 +216,12 @@ impl IPhysicalPlan for CommitSink {
                         self.deduplicated_label.clone(),
                         self.table_meta_timestamps,
                     )
-                })
+                })?;
+                attach_query_lineage_on_finished(
+                    &mut builder.main_pipeline,
+                    self.lineage_updates.clone(),
+                );
+                Ok(())
             }
         }
     }
diff --git a/src/query/service/src/physical_plans/physical_compact_source.rs b/src/query/service/src/physical_plans/physical_compact_source.rs
@@ -286,6 +286,7 @@ impl PhysicalPlanBuilder {
             deduplicated_label: None,
             recluster_info: None,
             table_meta_timestamps,
+            lineage_updates: vec![],
             meta: PhysicalPlanMeta::new("CommitSink"),
         });
 
diff --git a/src/query/service/src/physical_plans/physical_mutation.rs b/src/query/service/src/physical_plans/physical_mutation.rs
@@ -73,6 +73,7 @@ use tokio::sync::Semaphore;
 
 use super::ColumnMutation;
 use super::CommitType;
+use crate::interpreters::common::build_query_lineage_updates_with_lineage;
 use crate::physical_plans::CommitSink;
 use crate::physical_plans::Exchange;
 use crate::physical_plans::MutationManipulate;
@@ -320,6 +321,12 @@ impl PhysicalPlanBuilder {
 
         if *truncate_table {
             // Do truncate.
+            let lineage_updates = build_query_lineage_updates_with_lineage(
+                self.ctx.as_ref(),
+                self.query_lineage.as_deref(),
+                None,
+            )
+            .await?;
             plan = PhysicalPlan::new(CommitSink {
                 input: plan,
                 snapshot: mutation_build_info.table_snapshot,
@@ -333,6 +340,7 @@ impl PhysicalPlanBuilder {
                 recluster_info: None,
                 meta: PhysicalPlanMeta::new("CommitSink"),
                 table_meta_timestamps: mutation_build_info.table_meta_timestamps,
+                lineage_updates,
             });
             plan.adjust_plan_id(&mut 0);
             return Ok(plan);
@@ -412,6 +420,12 @@ impl PhysicalPlanBuilder {
                 });
             }
 
+            let lineage_updates = build_query_lineage_updates_with_lineage(
+                self.ctx.as_ref(),
+                self.query_lineage.as_deref(),
+                None,
+            )
+            .await?;
             plan = PhysicalPlan::new(CommitSink {
                 input: plan,
                 snapshot: mutation_build_info.table_snapshot,
@@ -426,6 +440,7 @@ impl PhysicalPlanBuilder {
                 meta: PhysicalPlanMeta::new("CommitSink"),
                 recluster_info: None,
                 table_meta_timestamps: mutation_build_info.table_meta_timestamps,
+                lineage_updates,
             });
 
             plan.adjust_plan_id(&mut 0);
@@ -653,6 +668,12 @@ impl PhysicalPlanBuilder {
         };
 
         // build mutation_aggregate
+        let lineage_updates = build_query_lineage_updates_with_lineage(
+            self.ctx.as_ref(),
+            self.query_lineage.as_deref(),
+            None,
+        )
+        .await?;
         let mut physical_plan = PhysicalPlan::new(CommitSink {
             input: plan,
             snapshot: mutation_build_info.table_snapshot,
@@ -667,6 +688,7 @@ impl PhysicalPlanBuilder {
             recluster_info: None,
             meta: PhysicalPlanMeta::new("CommitSink"),
             table_meta_timestamps: mutation_build_info.table_meta_timestamps,
+            lineage_updates,
         });
 
         physical_plan.adjust_plan_id(&mut 0);
diff --git a/src/query/service/src/physical_plans/physical_plan_builder.rs b/src/query/service/src/physical_plans/physical_plan_builder.rs
@@ -24,6 +24,7 @@ use databend_common_meta_app::schema::TableInfo;
 use databend_common_meta_app::schema::UpdateStreamMetaReq;
 use databend_common_sql::ColumnSet;
 use databend_common_sql::MetadataRef;
+use databend_common_sql::QueryLineage;
 use databend_common_sql::optimizer::ir::RelExpr;
 use databend_common_sql::optimizer::ir::SExpr;
 use databend_common_sql::plans::RelOperator;
@@ -44,6 +45,7 @@ pub struct PhysicalPlanBuilder {
     pub cte_required_columns: HashMap<String, ColumnSet>,
     pub is_cte_required_columns_collected: bool,
     pub build_depth: usize,
+    pub(crate) query_lineage: Option<Arc<QueryLineage>>,
 }
 
 impl PhysicalPlanBuilder {
@@ -58,6 +60,7 @@ impl PhysicalPlanBuilder {
             cte_required_columns: HashMap::new(),
             is_cte_required_columns_collected: false,
             build_depth: 0,
+            query_lineage: None,
         }
     }
 
@@ -190,6 +193,10 @@ impl PhysicalPlanBuilder {
         self.mutation_build_info = Some(mutation_build_info);
     }
 
+    pub(crate) fn set_query_lineage(&mut self, query_lineage: Option<Arc<QueryLineage>>) {
+        self.query_lineage = query_lineage;
+    }
+
     pub fn set_metadata(&mut self, metadata: MetadataRef) {
         self.metadata = metadata;
     }
diff --git a/src/query/service/src/sessions/query_ctx.rs b/src/query/service/src/sessions/query_ctx.rs
diff --git a/src/query/service/src/sessions/query_ctx_shared.rs b/src/query/service/src/sessions/query_ctx_shared.rs
diff --git a/src/query/service/tests/it/storages/fuse/operations/mutation/block_compact_mutator.rs b/src/query/service/tests/it/storages/fuse/operations/mutation/block_compact_mutator.rs

Original file line number	Diff line number	Diff line change
`@@ -440,6 +440,7 @@ impl ReclusterTableInterpreter {`
`440`	`440`	`deduplicated_label: None,`
`441`	`441`	`table_meta_timestamps,`
`442`	`442`	`recluster_info,`
	`443`	`+ lineage_updates: vec![],`
`443`	`444`	`meta: PhysicalPlanMeta::new("CommitSink"),`
`444`	`445`	`})`
`445`	`446`	`}`