feat(query): extract query lineage from plans

src/meta/app/src/schema/catalog.rs

@@ -26,7 +26,7 @@ use crate::schema::catalog_id_ident;
 use crate::storage::StorageParams;
 use crate::tenant::Tenant;
 
-#[derive(Clone, Copy, PartialEq, Eq, Debug, Hash)]
+#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Debug, Hash)]
 pub enum CatalogType {
     Default = 1,
     Hive = 2,

src/query/catalog/src/table.rs

@@ -109,6 +109,15 @@ pub trait Table: Sync + Send {
 
     fn get_table_info(&self) -> &TableInfo;
 
+    /// Returns the source table whose data columns a stream exposes.
+    ///
+    /// Lineage intentionally passes through a stream to its source table.
+    /// Views are lineage boundaries and must not use this relation-level hook;
+    /// their output columns are annotated separately by the planner.
+    fn stream_source_table_info(&self) -> Option<&TableInfo> {
+        None
+    }
+
     fn get_data_source_info(&self) -> DataSourceInfo {
         DataSourceInfo::TableSource(self.get_table_info().clone())
     }

src/query/service/src/interpreters/interpreter_table_create.rs

@@ -251,6 +251,12 @@ impl CreateTableInterpreter {
             overwrite: false,
             source: InsertInputSource::SelectPlan(select_plan),
             table_info: Some(table_info),
+            lineage_target_table_id: None,
+            lineage_target_catalog_type: if self.plan.engine == Engine::Iceberg {
+                databend_common_meta_app::schema::CatalogType::Iceberg
+            } else {
+                databend_common_meta_app::schema::CatalogType::Default
+            },
         };
 
         let mut pipeline = InsertInterpreter::try_create(self.ctx.clone(), insert_plan)?

src/query/sql/src/planner/binder/bind_table_reference/bind_table.rs

@@ -27,10 +27,13 @@ use databend_common_catalog::table_with_options::get_with_opt_consume;
 use databend_common_catalog::table_with_options::get_with_opt_max_batch_size;
 use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
+use databend_common_expression::ColumnId;
 use databend_common_storages_basic::view_table::QUERY;
 use databend_storages_common_table_meta::table::get_change_type;
 
 use crate::BindContext;
+use crate::LineageSourceRelation;
+use crate::ViewLineageSourceColumn;
 use crate::binder::Binder;
 use crate::binder::ViewIdent;
 use crate::binder::util::TableIdentifier;
@@ -189,17 +192,25 @@ impl Binder {
         {
             let change_type = get_change_type(&table_name_alias);
             if change_type.is_some() {
-                let table_index = self.metadata.write().add_table(
-                    catalog,
-                    database.clone(),
-                    table_meta.clone(),
-                    branch_name,
-                    table_name_alias,
-                    !bind_context.binding_views.is_empty(),
-                    bind_context.planning_agg_index,
-                    false,
-                    cte_suffix_name,
-                );
+                let stream_lineage_source = stream_lineage_source_relation(&table_meta);
+                let table_index = {
+                    let mut metadata = self.metadata.write();
+                    let table_index = metadata.add_table(
+                        catalog,
+                        database.clone(),
+                        table_meta.clone(),
+                        branch_name,
+                        table_name_alias,
+                        !bind_context.binding_views.is_empty(),
+                        bind_context.planning_agg_index,
+                        false,
+                        cte_suffix_name,
+                    );
+                    if let Some(stream_lineage_source) = stream_lineage_source {
+                        metadata.set_stream_lineage_source(table_index, stream_lineage_source);
+                    }
+                    table_index
+                };
                 let (s_expr, mut bind_context) = self.bind_base_table(
                     bind_context,
                     database.as_str(),
@@ -280,11 +291,11 @@ impl Binder {
                 new_bind_context.binding_views.insert(view_ident);
                 if let Statement::Query(query) = &stmt {
                     self.metadata.write().add_table(
-                        catalog,
+                        catalog.clone(),
                         database.clone(),
-                        table_meta,
-                        branch_name,
-                        table_name_alias,
+                        table_meta.clone(),
+                        branch_name.clone(),
+                        table_name_alias.clone(),
                         false,
                         false,
                         false,
@@ -302,6 +313,13 @@ impl Binder {
                             column.table_name = Some(self.normalize_identifier(table).name);
                         }
                     }
+                    self.add_view_lineage_source_columns(
+                        &new_bind_context,
+                        catalog.as_str(),
+                        database.as_str(),
+                        table_name.as_str(),
+                        &table_meta,
+                    );
                     // Restore binding_views to the outer scope's value so the
                     // current view does not leak into sibling/parent contexts.
                     new_bind_context.binding_views = bind_context.binding_views.clone();
@@ -343,4 +361,44 @@ impl Binder {
             }
         }
     }
+
+    fn add_view_lineage_source_columns(
+        &mut self,
+        bind_context: &BindContext,
+        catalog: &str,
+        database: &str,
+        view_name: &str,
+        view: &std::sync::Arc<dyn databend_common_catalog::table::Table>,
+    ) {
+        let relation = LineageSourceRelation {
+            catalog: catalog.to_string(),
+            database: database.to_string(),
+            name: view_name.to_string(),
+            id: view.get_table_info().ident.table_id,
+        };
+        let mut metadata = self.metadata.write();
+        for (idx, column) in bind_context.columns.iter().enumerate() {
+            metadata.add_view_lineage_source_column(column.index, ViewLineageSourceColumn {
+                relation: relation.clone(),
+                // View TableMeta has no persisted schema. The bound view query (including an
+                // explicit view column list) is the source of truth for output column names.
+                name: column.column_name.clone(),
+                id: idx as ColumnId,
+            });
+        }
+    }
+}
+
+fn stream_lineage_source_relation(
+    table: &std::sync::Arc<dyn databend_common_catalog::table::Table>,
+) -> Option<LineageSourceRelation> {
+    table.stream_source_table_info().and_then(|table_info| {
+        let database = table_info.database_name().ok()?.to_string();
+        Some(LineageSourceRelation {
+            catalog: table_info.catalog().to_string(),
+            database,
+            name: table_info.name.clone(),
+            id: table_info.ident.table_id,
+        })
+    })
 }

src/query/sql/src/planner/binder/ddl/view.rs

@@ -74,6 +74,7 @@ impl Binder {
             view_name,
             column_names,
             subquery,
+            query_plan: None,
         };
         Ok(Plan::CreateView(plan.into()))
     }

src/query/sql/src/planner/binder/insert.rs

@@ -311,6 +311,10 @@ impl Binder {
             overwrite: *overwrite,
             source: input_source?,
             table_info: None,
+            lineage_target_table_id: (table.get_table_info().catalog_info.catalog_type()
+                == databend_common_meta_app::schema::CatalogType::Default)
+                .then_some(table.get_table_info().ident.table_id),
+            lineage_target_catalog_type: table.get_table_info().catalog_info.catalog_type(),
         };
 
         Ok(Plan::Insert(Box::new(plan)))