You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@kyuubi.apache.org by GitBox <gi...@apache.org> on 2022/08/05 07:32:20 UTC
[GitHub] [incubator-kyuubi] iodone opened a new pull request, #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
iodone opened a new pull request, #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185
<!--
Thanks for sending a pull request!
Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://kyuubi.readthedocs.io/en/latest/community/contributions.html
2. If the PR is related to an issue in https://github.com/apache/incubator-kyuubi/issues, add '[KYUUBI #XXXX]' in your PR title, e.g., '[KYUUBI #XXXX] Your PR title ...'.
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][KYUUBI #XXXX] Your PR title ...'.
-->
### _Why are the changes needed?_
<!--
Please clarify why the changes are needed. For instance,
1. If you add a feature, you can talk about the use case of it.
2. If you fix a bug, you can clarify why it is a bug.
-->
`set kyuubi.operation.lineage.enabled = true` to enable lineage information logging. SQL lineage information will be written to `SparkOperationEvent` and output to the engine log. `kyuubi.operation.lineage.enabled `defaults to `false`.
Columns lineage definition
```
col0 -> (table.a, table.b)
col1 -> (table.c, table.a)
```
SQL lineage logger JSON format example.
**SQL:**
```
select a as col0, b as col1 from test_table0
```
**Lineage:**
```
{
"inputTables":[
"default.test_table0"
],
"outputTables":[
],
"columnLineage":[
[
"col0",
[
"default.test_table0.a"
]
],
[
"col1",
[
"default.test_table0.b"
]
]
]
}
```
Currently supported column lineage for spark `Command` and `Query` `TreeNode`:
### Query
- `Select`
### Command
- `CreateDataSourceTableAsSelectCommand`
- `CreateHiveTableAsSelectCommand`
- `OptimizedCreateHiveTableAsSelectCommand`
- `CreateTableAsSelect`
- `ReplaceTableAsSelect`
- `InsertIntoDataSourceCommand`
- `InsertIntoHadoopFsRelationCommand`
- `InsertIntoDataSourceDirCommand`
- `InsertIntoHiveDirCommand`
- `InsertIntoHiveTable`
### _How was this patch tested?_
- [ ] Add some test cases that check the changes thoroughly including negative and positive cases if possible
- [ ] Add screenshots for manual tests if appropriate
- [ ] [Run test](https://kyuubi.apache.org/docs/latest/develop_tools/testing.html#running-tests) locally before make a pull request
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] iodone commented on pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
iodone commented on PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#issuecomment-1217408709
> It seems Kyuubi can provide a way for Spark extensions to inject event to Kyuubi event, so we can always unify the Spark event into Kyuubi event. Maybe something like user can get KyuubiSparkSession by `KyuubiSparkSession.get` in Spark extension. What do you think ? @iodone @yaooqinn @pan3793 @turboFei
With this design, there are cases where the bottom layer depends on the top layer event？
> Let's clean the way 1 first, I think 2 is better and can still be improved
In some scenarios the user submits SQL only to get the lineage information and does not execute it. Just like `PlanOnlyMode`, we can output only the lineage information. In this case we have to implement the lineage extraction capability in Kyuubi Spark Engine. This implement can reuse the way 1.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] ulysses-you commented on a diff in pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
ulysses-you commented on code in PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#discussion_r944144310
##########
externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/spark/kyuubi/SparkSQLLineageParseHelper.scala:
##########
@@ -0,0 +1,309 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.kyuubi
+
+import scala.collection.immutable.ListMap
+import scala.util.{Failure, Success, Try}
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat, CatalogTable, HiveTableRelation}
+import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, AttributeSet, NamedExpression}
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.connector.catalog.{CatalogPlugin, Identifier, TableCatalog}
+import org.apache.spark.sql.execution.datasources.LogicalRelation
+import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, DataSourceV2ScanRelation}
+
+import org.apache.kyuubi.Logging
+import org.apache.kyuubi.engine.spark.KyuubiSparkUtil.isSparkVersionAtMost
+import org.apache.kyuubi.engine.spark.events.Lineage
+
+trait LineageParser {
+ def sparkSession: SparkSession
+
+ type AttributeMap[A] = ListMap[Attribute, A]
+
+ def parse(plan: LogicalPlan): Lineage = {
+ val columnsLineage =
+ extractColumnsLineage(plan, ListMap[Attribute, AttributeSet]()).toList.collect {
+ case (k, attrs) =>
+ k.name -> attrs.map(_.name).toSet
+ }
+ val (inputTables, outputTables) = columnsLineage.foldLeft((List[String](), List[String]())) {
+ case ((inputs, outputs), (out, in)) =>
+ val x = (inputs ++ in.map(_.split('.').init.mkString("."))).filter(_.nonEmpty)
+ val y = outputs ++ List(out.split('.').init.mkString(".")).filter(_.nonEmpty)
+ (x, y)
+ }
+ Lineage(inputTables.distinct, outputTables.distinct, columnsLineage)
+ }
+
+ private def mergeColumnsLineage(
+ left: AttributeMap[AttributeSet],
+ right: AttributeMap[AttributeSet]): AttributeMap[AttributeSet] = {
+ left ++ right.map {
+ case (k, attrs) =>
+ k -> (attrs ++ left.getOrElse(k, AttributeSet.empty))
+ }
+ }
+
+ private def joinColumnsLineage(
+ parent: AttributeMap[AttributeSet],
+ child: AttributeMap[AttributeSet]): AttributeMap[AttributeSet] = {
+ if (parent.isEmpty) child
+ else {
+ val childMap = child.map { case (k, attrs) => (k.exprId, attrs) }
+ parent.map { case (k, attrs) =>
+ k -> AttributeSet(attrs.flatMap(attr =>
+ childMap.getOrElse(attr.exprId, AttributeSet.empty)))
+ }
+ }
+ }
+
+ private def getSelectColumnLineage(
+ named: Seq[NamedExpression]): AttributeMap[AttributeSet] = {
+ val exps = named.map {
+ case a: Alias => a.toAttribute -> a.references
+ case a: Attribute => a -> a.references
+ }
+ ListMap(exps: _*)
+ }
+
+ private def joinRelationColumnLineage(
+ parent: AttributeMap[AttributeSet],
+ relationAttrs: Seq[Attribute],
+ tableName: String = ""): AttributeMap[AttributeSet] = {
+ val relationAttrSet = AttributeSet(relationAttrs)
+ if (parent.nonEmpty) {
+ parent.map { case (k, attrs) =>
+ k -> AttributeSet(attrs.collect {
+ case attr if relationAttrSet.contains(attr) =>
+ attr.withName(Seq(tableName, attr.name).filter(_.nonEmpty).mkString("."))
+ })
+ }
+ } else {
+ ListMap(relationAttrs.map { attr =>
+ (
+ attr,
+ AttributeSet(attr.withName(Seq(tableName, attr.name).filter(_.nonEmpty).mkString("."))))
+ }: _*)
+ }
+ }
+
+ private def extractColumnsLineage(
+ plan: LogicalPlan,
+ parentColumnsLineage: AttributeMap[AttributeSet]): AttributeMap[AttributeSet] = {
+
+ def getPlanField[T](field: String): T = {
+ getFieldVal[T](plan, field)
+ }
+
+ def getCurrentPlanField[T](curPlan: LogicalPlan, field: String): T = {
+ getFieldVal[T](curPlan, field)
+ }
+
+ def getPlanMethod[T](name: String): T = {
+ getMethod[T](plan, name)
+ }
+
+ def getQuery: LogicalPlan = {
+ getPlanField[LogicalPlan]("query")
+ }
+
+ plan match {
+ // For command
+ case p if p.nodeName == "AlterViewAsCommand" =>
+ val analyzed = sparkSession.sessionState.analyzer.execute(getQuery)
Review Comment:
why we need this ?
##########
externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/spark/kyuubi/SparkSQLLineageParseHelper.scala:
##########
@@ -0,0 +1,309 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.kyuubi
+
+import scala.collection.immutable.ListMap
+import scala.util.{Failure, Success, Try}
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat, CatalogTable, HiveTableRelation}
+import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, AttributeSet, NamedExpression}
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.connector.catalog.{CatalogPlugin, Identifier, TableCatalog}
+import org.apache.spark.sql.execution.datasources.LogicalRelation
+import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, DataSourceV2ScanRelation}
+
+import org.apache.kyuubi.Logging
+import org.apache.kyuubi.engine.spark.KyuubiSparkUtil.isSparkVersionAtMost
+import org.apache.kyuubi.engine.spark.events.Lineage
+
+trait LineageParser {
+ def sparkSession: SparkSession
+
+ type AttributeMap[A] = ListMap[Attribute, A]
+
+ def parse(plan: LogicalPlan): Lineage = {
+ val columnsLineage =
+ extractColumnsLineage(plan, ListMap[Attribute, AttributeSet]()).toList.collect {
+ case (k, attrs) =>
+ k.name -> attrs.map(_.name).toSet
+ }
+ val (inputTables, outputTables) = columnsLineage.foldLeft((List[String](), List[String]())) {
+ case ((inputs, outputs), (out, in)) =>
+ val x = (inputs ++ in.map(_.split('.').init.mkString("."))).filter(_.nonEmpty)
+ val y = outputs ++ List(out.split('.').init.mkString(".")).filter(_.nonEmpty)
+ (x, y)
+ }
+ Lineage(inputTables.distinct, outputTables.distinct, columnsLineage)
+ }
+
+ private def mergeColumnsLineage(
+ left: AttributeMap[AttributeSet],
+ right: AttributeMap[AttributeSet]): AttributeMap[AttributeSet] = {
+ left ++ right.map {
+ case (k, attrs) =>
+ k -> (attrs ++ left.getOrElse(k, AttributeSet.empty))
+ }
+ }
+
+ private def joinColumnsLineage(
+ parent: AttributeMap[AttributeSet],
+ child: AttributeMap[AttributeSet]): AttributeMap[AttributeSet] = {
+ if (parent.isEmpty) child
+ else {
+ val childMap = child.map { case (k, attrs) => (k.exprId, attrs) }
+ parent.map { case (k, attrs) =>
+ k -> AttributeSet(attrs.flatMap(attr =>
+ childMap.getOrElse(attr.exprId, AttributeSet.empty)))
+ }
+ }
+ }
+
+ private def getSelectColumnLineage(
+ named: Seq[NamedExpression]): AttributeMap[AttributeSet] = {
+ val exps = named.map {
+ case a: Alias => a.toAttribute -> a.references
+ case a: Attribute => a -> a.references
+ }
+ ListMap(exps: _*)
+ }
+
+ private def joinRelationColumnLineage(
+ parent: AttributeMap[AttributeSet],
+ relationAttrs: Seq[Attribute],
+ tableName: String = ""): AttributeMap[AttributeSet] = {
+ val relationAttrSet = AttributeSet(relationAttrs)
+ if (parent.nonEmpty) {
+ parent.map { case (k, attrs) =>
+ k -> AttributeSet(attrs.collect {
+ case attr if relationAttrSet.contains(attr) =>
+ attr.withName(Seq(tableName, attr.name).filter(_.nonEmpty).mkString("."))
+ })
+ }
+ } else {
+ ListMap(relationAttrs.map { attr =>
+ (
+ attr,
+ AttributeSet(attr.withName(Seq(tableName, attr.name).filter(_.nonEmpty).mkString("."))))
+ }: _*)
+ }
+ }
+
+ private def extractColumnsLineage(
+ plan: LogicalPlan,
+ parentColumnsLineage: AttributeMap[AttributeSet]): AttributeMap[AttributeSet] = {
+
+ def getPlanField[T](field: String): T = {
+ getFieldVal[T](plan, field)
+ }
+
+ def getCurrentPlanField[T](curPlan: LogicalPlan, field: String): T = {
+ getFieldVal[T](curPlan, field)
+ }
+
+ def getPlanMethod[T](name: String): T = {
+ getMethod[T](plan, name)
+ }
+
+ def getQuery: LogicalPlan = {
+ getPlanField[LogicalPlan]("query")
+ }
+
+ plan match {
+ // For command
+ case p if p.nodeName == "AlterViewAsCommand" =>
+ val analyzed = sparkSession.sessionState.analyzer.execute(getQuery)
+ val view = getPlanField[TableIdentifier]("name").unquotedString
+ extractColumnsLineage(analyzed, parentColumnsLineage).map { case (k, v) =>
+ k.withName(s"$view.${k.name}") -> v
+ }
+
+ case p if p.nodeName == "CreateViewCommand" =>
+ val view = getPlanField[TableIdentifier]("name").unquotedString
+ val outputCols =
+ getPlanField[Seq[(String, Option[String])]]("userSpecifiedColumns").map(_._1)
+ val query =
+ if (isSparkVersionAtMost("3.1")) {
+ getPlanField[LogicalPlan]("child")
+ } else {
+ getPlanField[LogicalPlan]("plan")
+ }
+ val analyzed = sparkSession.sessionState.analyzer.execute(query)
+ extractColumnsLineage(analyzed, parentColumnsLineage).zipWithIndex.map {
+ case ((k, v), i) if outputCols.nonEmpty => k.withName(s"$view.${outputCols(i)}") -> v
+ case ((k, v), _) => k.withName(s"$view.${k.name}") -> v
+ }
+
+ case p if p.nodeName == "CreateDataSourceTableAsSelectCommand" =>
+ val table = getPlanField[CatalogTable]("table").qualifiedName
+ val analyzed = sparkSession.sessionState.analyzer.execute(getQuery)
+ extractColumnsLineage(analyzed, parentColumnsLineage).map { case (k, v) =>
+ k.withName(s"$table.${k.name}") -> v
+ }
+
+ case p
+ if p.nodeName == "CreateHiveTableAsSelectCommand" ||
+ p.nodeName == "OptimizedCreateHiveTableAsSelectCommand" =>
+ val table = getPlanField[CatalogTable]("tableDesc").qualifiedName
+ extractColumnsLineage(getQuery, parentColumnsLineage).map { case (k, v) =>
+ k.withName(s"$table.${k.name}") -> v
+ }
+
+ case p
+ if p.nodeName == "CreateTableAsSelect" ||
+ p.nodeName == "ReplaceTableAsSelect" =>
+ val (table, namespace, catalog) =
+ if (isSparkVersionAtMost("3.2")) {
+ (
+ getPlanField[Identifier]("tableName").name,
+ getPlanField[Identifier]("tableName").namespace.mkString("."),
+ getPlanField[TableCatalog]("catalog").name())
+ } else {
+ (
+ getPlanMethod[Identifier]("tableName").name(),
+ getPlanMethod[Identifier]("tableName").namespace().mkString("."),
+ getCurrentPlanField[CatalogPlugin](
+ getPlanMethod[LogicalPlan]("left"),
+ "catalog").name())
+ }
+ extractColumnsLineage(getQuery, parentColumnsLineage).map { case (k, v) =>
+ k.withName(s"$catalog.$namespace.$table.${k.name}") -> v
+ }
+
+ case p if p.nodeName == "InsertIntoDataSourceCommand" =>
+ val logicalRelation = getPlanField[LogicalRelation]("logicalRelation")
+ val table = logicalRelation.catalogTable.map(_.qualifiedName).getOrElse("")
+ extractColumnsLineage(getQuery, parentColumnsLineage).map { case (k, v) =>
+ k.withName(s"$table.${k.name}") -> v
+ }
+
+ case p if p.nodeName == "InsertIntoHadoopFsRelationCommand" =>
+ val table =
+ getPlanField[Option[CatalogTable]]("catalogTable").map(_.qualifiedName).getOrElse("")
+ extractColumnsLineage(getQuery, parentColumnsLineage).map { case (k, v) =>
+ k.withName(s"$table.${k.name}") -> v
+ }
+
+ case p
+ if p.nodeName == "InsertIntoDataSourceDirCommand" ||
+ p.nodeName == "InsertIntoHiveDirCommand" =>
+ val dir =
+ getPlanField[CatalogStorageFormat]("storage").locationUri.map(_.toString).getOrElse("")
+ extractColumnsLineage(getQuery, parentColumnsLineage).map { case (k, v) =>
+ k.withName(s"`$dir`.${k.name}") -> v
+ }
+
+ case p if p.nodeName == "InsertIntoHiveTable" =>
+ val table = getPlanField[CatalogTable]("table").qualifiedName
+ extractColumnsLineage(getQuery, parentColumnsLineage).map { case (k, v) =>
+ k.withName(s"$table.${k.name}") -> v
+ }
+
+ case p if p.nodeName == "SaveIntoDataSourceCommand" =>
+ extractColumnsLineage(getQuery, parentColumnsLineage)
+
+ // For query
+ case p: Project =>
+ val nextColumnsLineage =
+ joinColumnsLineage(parentColumnsLineage, getSelectColumnLineage(p.projectList))
+ p.children.map(extractColumnsLineage(_, nextColumnsLineage)).reduce(mergeColumnsLineage)
+
+ case p: Aggregate =>
+ val nextColumnsLineage =
+ joinColumnsLineage(parentColumnsLineage, getSelectColumnLineage(p.aggregateExpressions))
+ p.children.map(extractColumnsLineage(_, nextColumnsLineage)).reduce(mergeColumnsLineage)
+
+ case p: Union =>
+ // merge all children in to one derivedColumns
+ val childrenUnion =
+ p.children.map(extractColumnsLineage(_, ListMap[Attribute, AttributeSet]())).map(
+ _.values).reduce {
+ (left, right) =>
+ left.zip(right).map(attr => attr._1 ++ attr._2)
+ }
+ val childrenColumnsLineage = ListMap(p.output.zip(childrenUnion): _*)
+ joinColumnsLineage(parentColumnsLineage, childrenColumnsLineage)
+
+ case p: LogicalRelation if p.catalogTable.nonEmpty =>
+ val tableName = p.catalogTable.get.qualifiedName
+ joinRelationColumnLineage(parentColumnsLineage, p.output, tableName)
+
+ case p: HiveTableRelation =>
+ val tableName = p.tableMeta.qualifiedName
+ joinRelationColumnLineage(parentColumnsLineage, p.output, tableName)
+
+ case p: DataSourceV2Relation =>
Review Comment:
For an optimzied plan, the v2 scan should always be `DataSourceV2ScanRelation`. We can remove this pattern
##########
externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/spark/kyuubi/SparkSQLLineageParseHelper.scala:
##########
@@ -0,0 +1,309 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.kyuubi
+
+import scala.collection.immutable.ListMap
+import scala.util.{Failure, Success, Try}
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat, CatalogTable, HiveTableRelation}
+import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, AttributeSet, NamedExpression}
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.connector.catalog.{CatalogPlugin, Identifier, TableCatalog}
+import org.apache.spark.sql.execution.datasources.LogicalRelation
+import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, DataSourceV2ScanRelation}
+
+import org.apache.kyuubi.Logging
+import org.apache.kyuubi.engine.spark.KyuubiSparkUtil.isSparkVersionAtMost
+import org.apache.kyuubi.engine.spark.events.Lineage
+
+trait LineageParser {
+ def sparkSession: SparkSession
+
+ type AttributeMap[A] = ListMap[Attribute, A]
+
+ def parse(plan: LogicalPlan): Lineage = {
+ val columnsLineage =
+ extractColumnsLineage(plan, ListMap[Attribute, AttributeSet]()).toList.collect {
+ case (k, attrs) =>
+ k.name -> attrs.map(_.name).toSet
+ }
+ val (inputTables, outputTables) = columnsLineage.foldLeft((List[String](), List[String]())) {
+ case ((inputs, outputs), (out, in)) =>
+ val x = (inputs ++ in.map(_.split('.').init.mkString("."))).filter(_.nonEmpty)
+ val y = outputs ++ List(out.split('.').init.mkString(".")).filter(_.nonEmpty)
+ (x, y)
+ }
+ Lineage(inputTables.distinct, outputTables.distinct, columnsLineage)
+ }
+
+ private def mergeColumnsLineage(
+ left: AttributeMap[AttributeSet],
+ right: AttributeMap[AttributeSet]): AttributeMap[AttributeSet] = {
+ left ++ right.map {
+ case (k, attrs) =>
+ k -> (attrs ++ left.getOrElse(k, AttributeSet.empty))
+ }
+ }
+
+ private def joinColumnsLineage(
+ parent: AttributeMap[AttributeSet],
+ child: AttributeMap[AttributeSet]): AttributeMap[AttributeSet] = {
+ if (parent.isEmpty) child
+ else {
+ val childMap = child.map { case (k, attrs) => (k.exprId, attrs) }
+ parent.map { case (k, attrs) =>
+ k -> AttributeSet(attrs.flatMap(attr =>
+ childMap.getOrElse(attr.exprId, AttributeSet.empty)))
+ }
+ }
+ }
+
+ private def getSelectColumnLineage(
+ named: Seq[NamedExpression]): AttributeMap[AttributeSet] = {
+ val exps = named.map {
+ case a: Alias => a.toAttribute -> a.references
+ case a: Attribute => a -> a.references
+ }
+ ListMap(exps: _*)
+ }
+
+ private def joinRelationColumnLineage(
+ parent: AttributeMap[AttributeSet],
+ relationAttrs: Seq[Attribute],
+ tableName: String = ""): AttributeMap[AttributeSet] = {
+ val relationAttrSet = AttributeSet(relationAttrs)
+ if (parent.nonEmpty) {
+ parent.map { case (k, attrs) =>
+ k -> AttributeSet(attrs.collect {
+ case attr if relationAttrSet.contains(attr) =>
+ attr.withName(Seq(tableName, attr.name).filter(_.nonEmpty).mkString("."))
+ })
+ }
+ } else {
+ ListMap(relationAttrs.map { attr =>
+ (
+ attr,
+ AttributeSet(attr.withName(Seq(tableName, attr.name).filter(_.nonEmpty).mkString("."))))
+ }: _*)
+ }
+ }
+
+ private def extractColumnsLineage(
+ plan: LogicalPlan,
+ parentColumnsLineage: AttributeMap[AttributeSet]): AttributeMap[AttributeSet] = {
+
+ def getPlanField[T](field: String): T = {
+ getFieldVal[T](plan, field)
+ }
+
+ def getCurrentPlanField[T](curPlan: LogicalPlan, field: String): T = {
+ getFieldVal[T](curPlan, field)
+ }
+
+ def getPlanMethod[T](name: String): T = {
+ getMethod[T](plan, name)
+ }
+
+ def getQuery: LogicalPlan = {
+ getPlanField[LogicalPlan]("query")
+ }
+
+ plan match {
+ // For command
+ case p if p.nodeName == "AlterViewAsCommand" =>
+ val analyzed = sparkSession.sessionState.analyzer.execute(getQuery)
+ val view = getPlanField[TableIdentifier]("name").unquotedString
+ extractColumnsLineage(analyzed, parentColumnsLineage).map { case (k, v) =>
+ k.withName(s"$view.${k.name}") -> v
+ }
+
+ case p if p.nodeName == "CreateViewCommand" =>
+ val view = getPlanField[TableIdentifier]("name").unquotedString
+ val outputCols =
+ getPlanField[Seq[(String, Option[String])]]("userSpecifiedColumns").map(_._1)
+ val query =
+ if (isSparkVersionAtMost("3.1")) {
+ getPlanField[LogicalPlan]("child")
+ } else {
+ getPlanField[LogicalPlan]("plan")
+ }
+ val analyzed = sparkSession.sessionState.analyzer.execute(query)
+ extractColumnsLineage(analyzed, parentColumnsLineage).zipWithIndex.map {
+ case ((k, v), i) if outputCols.nonEmpty => k.withName(s"$view.${outputCols(i)}") -> v
+ case ((k, v), _) => k.withName(s"$view.${k.name}") -> v
+ }
+
+ case p if p.nodeName == "CreateDataSourceTableAsSelectCommand" =>
+ val table = getPlanField[CatalogTable]("table").qualifiedName
+ val analyzed = sparkSession.sessionState.analyzer.execute(getQuery)
+ extractColumnsLineage(analyzed, parentColumnsLineage).map { case (k, v) =>
+ k.withName(s"$table.${k.name}") -> v
+ }
+
+ case p
+ if p.nodeName == "CreateHiveTableAsSelectCommand" ||
+ p.nodeName == "OptimizedCreateHiveTableAsSelectCommand" =>
+ val table = getPlanField[CatalogTable]("tableDesc").qualifiedName
+ extractColumnsLineage(getQuery, parentColumnsLineage).map { case (k, v) =>
+ k.withName(s"$table.${k.name}") -> v
+ }
+
+ case p
+ if p.nodeName == "CreateTableAsSelect" ||
+ p.nodeName == "ReplaceTableAsSelect" =>
+ val (table, namespace, catalog) =
+ if (isSparkVersionAtMost("3.2")) {
+ (
+ getPlanField[Identifier]("tableName").name,
+ getPlanField[Identifier]("tableName").namespace.mkString("."),
+ getPlanField[TableCatalog]("catalog").name())
+ } else {
+ (
+ getPlanMethod[Identifier]("tableName").name(),
+ getPlanMethod[Identifier]("tableName").namespace().mkString("."),
+ getCurrentPlanField[CatalogPlugin](
+ getPlanMethod[LogicalPlan]("left"),
+ "catalog").name())
+ }
+ extractColumnsLineage(getQuery, parentColumnsLineage).map { case (k, v) =>
+ k.withName(s"$catalog.$namespace.$table.${k.name}") -> v
+ }
+
+ case p if p.nodeName == "InsertIntoDataSourceCommand" =>
+ val logicalRelation = getPlanField[LogicalRelation]("logicalRelation")
+ val table = logicalRelation.catalogTable.map(_.qualifiedName).getOrElse("")
+ extractColumnsLineage(getQuery, parentColumnsLineage).map { case (k, v) =>
+ k.withName(s"$table.${k.name}") -> v
+ }
+
+ case p if p.nodeName == "InsertIntoHadoopFsRelationCommand" =>
+ val table =
+ getPlanField[Option[CatalogTable]]("catalogTable").map(_.qualifiedName).getOrElse("")
+ extractColumnsLineage(getQuery, parentColumnsLineage).map { case (k, v) =>
+ k.withName(s"$table.${k.name}") -> v
+ }
+
+ case p
+ if p.nodeName == "InsertIntoDataSourceDirCommand" ||
+ p.nodeName == "InsertIntoHiveDirCommand" =>
+ val dir =
+ getPlanField[CatalogStorageFormat]("storage").locationUri.map(_.toString).getOrElse("")
+ extractColumnsLineage(getQuery, parentColumnsLineage).map { case (k, v) =>
+ k.withName(s"`$dir`.${k.name}") -> v
+ }
+
+ case p if p.nodeName == "InsertIntoHiveTable" =>
+ val table = getPlanField[CatalogTable]("table").qualifiedName
+ extractColumnsLineage(getQuery, parentColumnsLineage).map { case (k, v) =>
+ k.withName(s"$table.${k.name}") -> v
+ }
+
+ case p if p.nodeName == "SaveIntoDataSourceCommand" =>
+ extractColumnsLineage(getQuery, parentColumnsLineage)
+
+ // For query
+ case p: Project =>
+ val nextColumnsLineage =
+ joinColumnsLineage(parentColumnsLineage, getSelectColumnLineage(p.projectList))
+ p.children.map(extractColumnsLineage(_, nextColumnsLineage)).reduce(mergeColumnsLineage)
+
+ case p: Aggregate =>
+ val nextColumnsLineage =
+ joinColumnsLineage(parentColumnsLineage, getSelectColumnLineage(p.aggregateExpressions))
+ p.children.map(extractColumnsLineage(_, nextColumnsLineage)).reduce(mergeColumnsLineage)
+
+ case p: Union =>
+ // merge all children in to one derivedColumns
+ val childrenUnion =
+ p.children.map(extractColumnsLineage(_, ListMap[Attribute, AttributeSet]())).map(
+ _.values).reduce {
+ (left, right) =>
+ left.zip(right).map(attr => attr._1 ++ attr._2)
+ }
+ val childrenColumnsLineage = ListMap(p.output.zip(childrenUnion): _*)
+ joinColumnsLineage(parentColumnsLineage, childrenColumnsLineage)
+
+ case p: LogicalRelation if p.catalogTable.nonEmpty =>
+ val tableName = p.catalogTable.get.qualifiedName
+ joinRelationColumnLineage(parentColumnsLineage, p.output, tableName)
+
+ case p: HiveTableRelation =>
+ val tableName = p.tableMeta.qualifiedName
+ joinRelationColumnLineage(parentColumnsLineage, p.output, tableName)
+
+ case p: DataSourceV2Relation =>
+ val tableName = p.name
+ joinRelationColumnLineage(parentColumnsLineage, p.output, tableName)
+
+ case p: DataSourceV2ScanRelation =>
+ val tableName = p.name
+ joinRelationColumnLineage(parentColumnsLineage, p.output, tableName)
+
+ case p: LocalRelation =>
+ joinRelationColumnLineage(parentColumnsLineage, p.output)
Review Comment:
tableName = `__local__` ? so we can remove the default empty tableName
##########
externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/spark/kyuubi/SparkSQLLineageParseHelper.scala:
##########
@@ -0,0 +1,309 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.kyuubi
+
+import scala.collection.immutable.ListMap
+import scala.util.{Failure, Success, Try}
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat, CatalogTable, HiveTableRelation}
+import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, AttributeSet, NamedExpression}
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.connector.catalog.{CatalogPlugin, Identifier, TableCatalog}
+import org.apache.spark.sql.execution.datasources.LogicalRelation
+import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, DataSourceV2ScanRelation}
+
+import org.apache.kyuubi.Logging
+import org.apache.kyuubi.engine.spark.KyuubiSparkUtil.isSparkVersionAtMost
+import org.apache.kyuubi.engine.spark.events.Lineage
+
+trait LineageParser {
+ def sparkSession: SparkSession
+
+ type AttributeMap[A] = ListMap[Attribute, A]
+
+ def parse(plan: LogicalPlan): Lineage = {
+ val columnsLineage =
+ extractColumnsLineage(plan, ListMap[Attribute, AttributeSet]()).toList.collect {
+ case (k, attrs) =>
+ k.name -> attrs.map(_.name).toSet
+ }
+ val (inputTables, outputTables) = columnsLineage.foldLeft((List[String](), List[String]())) {
+ case ((inputs, outputs), (out, in)) =>
+ val x = (inputs ++ in.map(_.split('.').init.mkString("."))).filter(_.nonEmpty)
+ val y = outputs ++ List(out.split('.').init.mkString(".")).filter(_.nonEmpty)
+ (x, y)
+ }
+ Lineage(inputTables.distinct, outputTables.distinct, columnsLineage)
+ }
+
+ private def mergeColumnsLineage(
+ left: AttributeMap[AttributeSet],
+ right: AttributeMap[AttributeSet]): AttributeMap[AttributeSet] = {
+ left ++ right.map {
+ case (k, attrs) =>
+ k -> (attrs ++ left.getOrElse(k, AttributeSet.empty))
+ }
+ }
+
+ private def joinColumnsLineage(
+ parent: AttributeMap[AttributeSet],
+ child: AttributeMap[AttributeSet]): AttributeMap[AttributeSet] = {
+ if (parent.isEmpty) child
+ else {
+ val childMap = child.map { case (k, attrs) => (k.exprId, attrs) }
+ parent.map { case (k, attrs) =>
+ k -> AttributeSet(attrs.flatMap(attr =>
+ childMap.getOrElse(attr.exprId, AttributeSet.empty)))
+ }
+ }
+ }
+
+ private def getSelectColumnLineage(
+ named: Seq[NamedExpression]): AttributeMap[AttributeSet] = {
+ val exps = named.map {
+ case a: Alias => a.toAttribute -> a.references
+ case a: Attribute => a -> a.references
+ }
+ ListMap(exps: _*)
+ }
+
+ private def joinRelationColumnLineage(
+ parent: AttributeMap[AttributeSet],
+ relationAttrs: Seq[Attribute],
+ tableName: String = ""): AttributeMap[AttributeSet] = {
+ val relationAttrSet = AttributeSet(relationAttrs)
+ if (parent.nonEmpty) {
+ parent.map { case (k, attrs) =>
+ k -> AttributeSet(attrs.collect {
+ case attr if relationAttrSet.contains(attr) =>
+ attr.withName(Seq(tableName, attr.name).filter(_.nonEmpty).mkString("."))
+ })
+ }
+ } else {
+ ListMap(relationAttrs.map { attr =>
+ (
+ attr,
+ AttributeSet(attr.withName(Seq(tableName, attr.name).filter(_.nonEmpty).mkString("."))))
+ }: _*)
+ }
+ }
+
+ private def extractColumnsLineage(
+ plan: LogicalPlan,
+ parentColumnsLineage: AttributeMap[AttributeSet]): AttributeMap[AttributeSet] = {
+
+ def getPlanField[T](field: String): T = {
+ getFieldVal[T](plan, field)
+ }
+
+ def getCurrentPlanField[T](curPlan: LogicalPlan, field: String): T = {
+ getFieldVal[T](curPlan, field)
+ }
+
+ def getPlanMethod[T](name: String): T = {
+ getMethod[T](plan, name)
+ }
+
+ def getQuery: LogicalPlan = {
+ getPlanField[LogicalPlan]("query")
+ }
+
+ plan match {
+ // For command
+ case p if p.nodeName == "AlterViewAsCommand" =>
+ val analyzed = sparkSession.sessionState.analyzer.execute(getQuery)
+ val view = getPlanField[TableIdentifier]("name").unquotedString
+ extractColumnsLineage(analyzed, parentColumnsLineage).map { case (k, v) =>
+ k.withName(s"$view.${k.name}") -> v
+ }
+
+ case p if p.nodeName == "CreateViewCommand" =>
+ val view = getPlanField[TableIdentifier]("name").unquotedString
+ val outputCols =
+ getPlanField[Seq[(String, Option[String])]]("userSpecifiedColumns").map(_._1)
+ val query =
+ if (isSparkVersionAtMost("3.1")) {
+ getPlanField[LogicalPlan]("child")
+ } else {
+ getPlanField[LogicalPlan]("plan")
+ }
+ val analyzed = sparkSession.sessionState.analyzer.execute(query)
+ extractColumnsLineage(analyzed, parentColumnsLineage).zipWithIndex.map {
+ case ((k, v), i) if outputCols.nonEmpty => k.withName(s"$view.${outputCols(i)}") -> v
+ case ((k, v), _) => k.withName(s"$view.${k.name}") -> v
+ }
+
+ case p if p.nodeName == "CreateDataSourceTableAsSelectCommand" =>
+ val table = getPlanField[CatalogTable]("table").qualifiedName
+ val analyzed = sparkSession.sessionState.analyzer.execute(getQuery)
+ extractColumnsLineage(analyzed, parentColumnsLineage).map { case (k, v) =>
+ k.withName(s"$table.${k.name}") -> v
+ }
+
+ case p
+ if p.nodeName == "CreateHiveTableAsSelectCommand" ||
+ p.nodeName == "OptimizedCreateHiveTableAsSelectCommand" =>
+ val table = getPlanField[CatalogTable]("tableDesc").qualifiedName
+ extractColumnsLineage(getQuery, parentColumnsLineage).map { case (k, v) =>
+ k.withName(s"$table.${k.name}") -> v
+ }
+
+ case p
+ if p.nodeName == "CreateTableAsSelect" ||
+ p.nodeName == "ReplaceTableAsSelect" =>
+ val (table, namespace, catalog) =
+ if (isSparkVersionAtMost("3.2")) {
+ (
+ getPlanField[Identifier]("tableName").name,
+ getPlanField[Identifier]("tableName").namespace.mkString("."),
+ getPlanField[TableCatalog]("catalog").name())
+ } else {
+ (
+ getPlanMethod[Identifier]("tableName").name(),
+ getPlanMethod[Identifier]("tableName").namespace().mkString("."),
+ getCurrentPlanField[CatalogPlugin](
+ getPlanMethod[LogicalPlan]("left"),
+ "catalog").name())
+ }
+ extractColumnsLineage(getQuery, parentColumnsLineage).map { case (k, v) =>
+ k.withName(s"$catalog.$namespace.$table.${k.name}") -> v
+ }
+
+ case p if p.nodeName == "InsertIntoDataSourceCommand" =>
+ val logicalRelation = getPlanField[LogicalRelation]("logicalRelation")
+ val table = logicalRelation.catalogTable.map(_.qualifiedName).getOrElse("")
+ extractColumnsLineage(getQuery, parentColumnsLineage).map { case (k, v) =>
+ k.withName(s"$table.${k.name}") -> v
Review Comment:
the `.` of `$table.` should be merged to `$table` in case the table is empty
##########
externals/kyuubi-spark-sql-engine/src/test/scala/org/apache/spark/kyuubi/SparkSQLLineageParserHelperSuite.scala:
##########
@@ -0,0 +1,690 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.kyuubi
+
+import scala.collection.immutable.List
+import scala.reflect.io.File
+
+import org.apache.spark.sql.{DataFrame, SparkSession, SQLContext}
+import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat, CatalogTable, CatalogTableType}
+import org.apache.spark.sql.sources.{BaseRelation, InsertableRelation, SchemaRelationProvider}
+import org.apache.spark.sql.types.{IntegerType, StringType, StructType}
+
+import org.apache.kyuubi.KyuubiFunSuite
+import org.apache.kyuubi.engine.spark.KyuubiSparkUtil.isSparkVersionAtMost
+import org.apache.kyuubi.engine.spark.WithSparkSQLEngine
+import org.apache.kyuubi.engine.spark.events.Lineage
+
+class SparkSQLLineageParserHelperSuite extends WithSparkSQLEngine with KyuubiFunSuite {
+ val catalogName =
+ if (isSparkVersionAtMost("3.1")) "org.apache.spark.sql.connector.InMemoryTableCatalog"
+ else "org.apache.spark.sql.connector.catalog.InMemoryTableCatalog"
+
+ override def withKyuubiConf: Map[String, String] = Map(
+ "spark.sql.catalog.v2_catalog" -> catalogName)
+
+ override def beforeAll(): Unit = {
+ super.beforeAll()
+ spark.sql(s"create database if not exists test_db0")
+ spark.sql(s"create table if not exists test_db0.test_table0" +
+ s" (key int, value string) using parquet")
+ spark.sql(s"create table if not exists test_db0.test_table_part0" +
+ s" (key int, value string, pid string) using parquet" +
+ s" partitioned by(pid)")
+ }
+
+ override def afterAll(): Unit = {
+ Seq("test_db0.test_table0", "test_db0.test_table_part0").foreach { t =>
+ spark.sql(s"drop table if exists $t")
+ }
+ spark.sql(s"drop database if exists test_db0")
+ spark.stop()
+ super.afterAll()
+ }
+
+ protected def withTable(t: String)(f: String => Unit): Unit = {
+ try {
+ f(t)
+ } finally {
+ spark.sql(s"DROP TABLE IF EXISTS $t")
+ }
+ }
+
+ test("columns lineage extract - AlterViewAsCommand") {
+ spark.sql(s"create view alterviewascommand as select key from test_db0.test_table0")
Review Comment:
```scala
withTempView("") {
...
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] pan3793 commented on a diff in pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
pan3793 commented on code in PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#discussion_r952410705
##########
docs/extensions/engines/spark/lineage.md:
##########
@@ -0,0 +1,167 @@
+<!--
+ - Licensed to the Apache Software Foundation (ASF) under one or more
+ - contributor license agreements. See the NOTICE file distributed with
+ - this work for additional information regarding copyright ownership.
+ - The ASF licenses this file to You under the Apache License, Version 2.0
+ - (the "License"); you may not use this file except in compliance with
+ - the License. You may obtain a copy of the License at
+ -
+ - http://www.apache.org/licenses/LICENSE-2.0
+ -
+ - Unless required by applicable law or agreed to in writing, software
+ - distributed under the License is distributed on an "AS IS" BASIS,
+ - WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ - See the License for the specific language governing permissions and
+ - limitations under the License.
+ -->
+
+# SQL Lineage Support
+
+When we execute a SQL statement, the computing engine will return to a result set,
+each column in the result set may originate from different tables, and these different
+tables depend on other tables, which may have been calculated by functions, aggregation, etc.
+The source table is related to the result set, which is the SQL Lineage.
+
+## Introduction
+
+The current lineage parsing functionality is implemented as a plugin by extending Spark's `QueryExecutionListener`.
+1. The `SparkListenerSQLExecutionEnd` event is triggered after the SQL execution is finished and captured by the `QueryExecuctionListener`,
+ where the SQL lineage parsing process is performed on the successfully executed SQL.
+2. Will write the parsed lineage information to the log file in JSON format.
+
+### Example
+
+When the following SQL is executed:
+```sql
+## table
+create table test_table0(a string, b string)
+
+## query
+select a as col0, b as col1 from test_table0
+```
+The lineage of this SQL:
+```json
+{
+ "inputTables": ["default.test_table0"],
+ "outputTables": [],
+ "columnLineage": [{
+ "column": "col0",
+ "ordinal": 0,
Review Comment:
`ordinal` is redundant if `columnLineage` is a array
##########
docs/extensions/engines/spark/lineage.md:
##########
@@ -0,0 +1,167 @@
+<!--
+ - Licensed to the Apache Software Foundation (ASF) under one or more
+ - contributor license agreements. See the NOTICE file distributed with
+ - this work for additional information regarding copyright ownership.
+ - The ASF licenses this file to You under the Apache License, Version 2.0
+ - (the "License"); you may not use this file except in compliance with
+ - the License. You may obtain a copy of the License at
+ -
+ - http://www.apache.org/licenses/LICENSE-2.0
+ -
+ - Unless required by applicable law or agreed to in writing, software
+ - distributed under the License is distributed on an "AS IS" BASIS,
+ - WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ - See the License for the specific language governing permissions and
+ - limitations under the License.
+ -->
+
+# SQL Lineage Support
+
+When we execute a SQL statement, the computing engine will return to a result set,
+each column in the result set may originate from different tables, and these different
+tables depend on other tables, which may have been calculated by functions, aggregation, etc.
+The source table is related to the result set, which is the SQL Lineage.
+
+## Introduction
+
+The current lineage parsing functionality is implemented as a plugin by extending Spark's `QueryExecutionListener`.
+1. The `SparkListenerSQLExecutionEnd` event is triggered after the SQL execution is finished and captured by the `QueryExecuctionListener`,
+ where the SQL lineage parsing process is performed on the successfully executed SQL.
+2. Will write the parsed lineage information to the log file in JSON format.
+
+### Example
+
+When the following SQL is executed:
+```sql
+## table
+create table test_table0(a string, b string)
+
+## query
+select a as col0, b as col1 from test_table0
+```
+The lineage of this SQL:
+```json
+{
+ "inputTables": ["default.test_table0"],
+ "outputTables": [],
+ "columnLineage": [{
+ "column": "col0",
+ "ordinal": 0,
Review Comment:
`ordinal` is redundant if `columnLineage` is an array
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] yaooqinn closed pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
yaooqinn closed pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
URL: https://github.com/apache/incubator-kyuubi/pull/3185
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] pan3793 commented on a diff in pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
pan3793 commented on code in PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#discussion_r951033383
##########
pom.xml:
##########
@@ -67,6 +67,7 @@
<modules>
<module>dev/kyuubi-codecov</module>
<module>extensions/server/kyuubi-server-plugin</module>
+ <module>extensions/spark/kyuubi-spark-listener</module>
Review Comment:
Sort alphabetically
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] iodone commented on a diff in pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
iodone commented on code in PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#discussion_r947429110
##########
extensions/spark/kyuubi-spark-listener/src/main/scala/org/apache/kyuubi/events/LineageEventHandlerRegister.scala:
##########
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.events
+
+import org.apache.spark.sql.SparkSession
+
+import org.apache.kyuubi.SparkListenerSQLConf
+
+object LineageEventHandlerRegister {
+
+ @volatile var isRegister: Boolean = false
+
+ def register(spark: SparkSession): Unit = {
+ if (!isRegister) {
+ val logPath =
+ spark.conf.getOption(SparkListenerSQLConf.LINEAGE_EVENT_JSON_LOG_PATH.key).getOrElse(
+ SparkListenerSQLConf.LINEAGE_EVENT_JSON_LOG_PATH.defaultValStr)
+ val handler = LineageJsonLoggingEventHandler(
+ spark.sparkContext.applicationAttemptId
+ .map(id => s"${spark.sparkContext.applicationId}_$id")
+ .getOrElse(spark.sparkContext.applicationId),
+ spark.sparkContext.hadoopConfiguration,
+ logPath)
+ EventBus.register[OperationLineageEvent](handler)
Review Comment:
Very much agree with you.
I think Lineage Plugin can have its own output logger, which can write files or send to other external systems such as kafka, and the kyuubi spark engine side can also be configured with handlers to capture this event for other processing.
This is an example of how the plugin output of lineage information is implemented.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] iodone commented on a diff in pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
iodone commented on code in PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#discussion_r945037904
##########
externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/spark/kyuubi/SparkSQLLineageParseHelper.scala:
##########
@@ -0,0 +1,309 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.kyuubi
+
+import scala.collection.immutable.ListMap
+import scala.util.{Failure, Success, Try}
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat, CatalogTable, HiveTableRelation}
+import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, AttributeSet, NamedExpression}
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.connector.catalog.{CatalogPlugin, Identifier, TableCatalog}
+import org.apache.spark.sql.execution.datasources.LogicalRelation
+import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, DataSourceV2ScanRelation}
+
+import org.apache.kyuubi.Logging
+import org.apache.kyuubi.engine.spark.KyuubiSparkUtil.isSparkVersionAtMost
+import org.apache.kyuubi.engine.spark.events.Lineage
+
+trait LineageParser {
+ def sparkSession: SparkSession
+
+ type AttributeMap[A] = ListMap[Attribute, A]
+
+ def parse(plan: LogicalPlan): Lineage = {
+ val columnsLineage =
+ extractColumnsLineage(plan, ListMap[Attribute, AttributeSet]()).toList.collect {
+ case (k, attrs) =>
+ k.name -> attrs.map(_.name).toSet
+ }
+ val (inputTables, outputTables) = columnsLineage.foldLeft((List[String](), List[String]())) {
+ case ((inputs, outputs), (out, in)) =>
+ val x = (inputs ++ in.map(_.split('.').init.mkString("."))).filter(_.nonEmpty)
+ val y = outputs ++ List(out.split('.').init.mkString(".")).filter(_.nonEmpty)
+ (x, y)
+ }
+ Lineage(inputTables.distinct, outputTables.distinct, columnsLineage)
+ }
+
+ private def mergeColumnsLineage(
+ left: AttributeMap[AttributeSet],
+ right: AttributeMap[AttributeSet]): AttributeMap[AttributeSet] = {
+ left ++ right.map {
+ case (k, attrs) =>
+ k -> (attrs ++ left.getOrElse(k, AttributeSet.empty))
+ }
+ }
+
+ private def joinColumnsLineage(
+ parent: AttributeMap[AttributeSet],
+ child: AttributeMap[AttributeSet]): AttributeMap[AttributeSet] = {
+ if (parent.isEmpty) child
+ else {
+ val childMap = child.map { case (k, attrs) => (k.exprId, attrs) }
+ parent.map { case (k, attrs) =>
+ k -> AttributeSet(attrs.flatMap(attr =>
+ childMap.getOrElse(attr.exprId, AttributeSet.empty)))
+ }
+ }
+ }
+
+ private def getSelectColumnLineage(
+ named: Seq[NamedExpression]): AttributeMap[AttributeSet] = {
+ val exps = named.map {
+ case a: Alias => a.toAttribute -> a.references
+ case a: Attribute => a -> a.references
+ }
+ ListMap(exps: _*)
+ }
+
+ private def joinRelationColumnLineage(
+ parent: AttributeMap[AttributeSet],
+ relationAttrs: Seq[Attribute],
+ tableName: String = ""): AttributeMap[AttributeSet] = {
+ val relationAttrSet = AttributeSet(relationAttrs)
+ if (parent.nonEmpty) {
+ parent.map { case (k, attrs) =>
+ k -> AttributeSet(attrs.collect {
+ case attr if relationAttrSet.contains(attr) =>
+ attr.withName(Seq(tableName, attr.name).filter(_.nonEmpty).mkString("."))
+ })
+ }
+ } else {
+ ListMap(relationAttrs.map { attr =>
+ (
+ attr,
+ AttributeSet(attr.withName(Seq(tableName, attr.name).filter(_.nonEmpty).mkString("."))))
+ }: _*)
+ }
+ }
+
+ private def extractColumnsLineage(
+ plan: LogicalPlan,
+ parentColumnsLineage: AttributeMap[AttributeSet]): AttributeMap[AttributeSet] = {
+
+ def getPlanField[T](field: String): T = {
+ getFieldVal[T](plan, field)
+ }
+
+ def getCurrentPlanField[T](curPlan: LogicalPlan, field: String): T = {
+ getFieldVal[T](curPlan, field)
+ }
+
+ def getPlanMethod[T](name: String): T = {
+ getMethod[T](plan, name)
+ }
+
+ def getQuery: LogicalPlan = {
+ getPlanField[LogicalPlan]("query")
+ }
+
+ plan match {
+ // For command
+ case p if p.nodeName == "AlterViewAsCommand" =>
+ val analyzed = sparkSession.sessionState.analyzer.execute(getQuery)
Review Comment:
In Spark3.1, the subquery plan of `AlterViewAsCommand` is not unresolved. In addition, Spark 3.2/3.3 has been resolved
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] yaooqinn commented on pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
yaooqinn commented on PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#issuecomment-1212846274
also, this feature needs a complete usage doc
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] cfmcgrady commented on a diff in pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
cfmcgrady commented on code in PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#discussion_r952337414
##########
extensions/spark/kyuubi-spark-lineage/src/test/scala/org/apache/kyuubi/plugin/lineage/helper/SparkSQLLineageParserHelperSuite.scala:
##########
@@ -0,0 +1,722 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.lineage.helper
+
+import scala.collection.immutable.List
+import scala.reflect.io.File
+
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.{DataFrame, SparkListenerExtensionTest, SparkSession, SQLContext}
+import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat, CatalogTable, CatalogTableType}
+import org.apache.spark.sql.sources.{BaseRelation, InsertableRelation, SchemaRelationProvider}
+import org.apache.spark.sql.types.{IntegerType, StringType, StructType}
+
+import org.apache.kyuubi.KyuubiFunSuite
+import org.apache.kyuubi.plugin.lineage.events.Lineage
+import org.apache.kyuubi.plugin.lineage.helper.SparkListenerHelper.isSparkVersionAtMost
+
+class SparkSQLLineageParserHelperSuite extends KyuubiFunSuite
+ with SparkListenerExtensionTest {
+
+ val catalogName =
+ if (isSparkVersionAtMost("3.1")) "org.apache.spark.sql.connector.InMemoryTableCatalog"
+ else "org.apache.spark.sql.connector.catalog.InMemoryTableCatalog"
+
+ override protected val catalogImpl: String = "hive"
+
+ override def sparkConf(): SparkConf = {
+ super.sparkConf()
+ .set(
+ "spark.sql.catalog.v2_catalog",
+ catalogName)
+ }
+
+ override def beforeAll(): Unit = {
+ super.beforeAll()
+ spark.sql(s"create database if not exists test_db")
+ spark.sql(s"create database if not exists test_db0")
+ spark.sql(s"create table if not exists test_db0.test_table0" +
+ s" (key int, value string) using parquet")
+ spark.sql(s"create table if not exists test_db0.test_table_part0" +
+ s" (key int, value string, pid string) using parquet" +
Review Comment:
Should we test both `hive` and `parquet`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] ulysses-you commented on a diff in pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
ulysses-you commented on code in PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#discussion_r951025424
##########
extensions/spark/kyuubi-spark-listener/pom.xml:
##########
@@ -0,0 +1,103 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ Licensed to the Apache Software Foundation (ASF) under one or more
+ ~ contributor license agreements. See the NOTICE file distributed with
+ ~ this work for additional information regarding copyright ownership.
+ ~ The ASF licenses this file to You under the Apache License, Version 2.0
+ ~ (the "License"); you may not use this file except in compliance with
+ ~ the License. You may obtain a copy of the License at
+ ~
+ ~ http://www.apache.org/licenses/LICENSE-2.0
+ ~
+ ~ Unless required by applicable law or agreed to in writing, software
+ ~ distributed under the License is distributed on an "AS IS" BASIS,
+ ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ ~ See the License for the specific language governing permissions and
+ ~ limitations under the License.
+ -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <parent>
+ <artifactId>kyuubi-parent</artifactId>
+ <groupId>org.apache.kyuubi</groupId>
+ <version>1.7.0-SNAPSHOT</version>
+ <relativePath>../../../pom.xml</relativePath>
+
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+
+ <artifactId>kyuubi-spark-listener_2.12</artifactId>
Review Comment:
`kyuubi-spark-lineage_2.12`
##########
extensions/spark/kyuubi-spark-listener/src/main/scala/org/apache/kyuubi/plugin/lineage/helper/SparkListenerHelper.scala:
##########
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.lineage.helper
+
+import org.apache.spark.SPARK_VERSION
+
+import org.apache.kyuubi.Logging
+import org.apache.kyuubi.engine.SemanticVersion
+
+object SparkListenerHelper extends Logging {
Review Comment:
unused: `Logging`
##########
extensions/spark/kyuubi-spark-listener/src/main/scala/org/apache/kyuubi/plugin/lineage/helper/SparkSQLLineageParseHelper.scala:
##########
@@ -0,0 +1,311 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.lineage.helper
+
+import scala.collection.immutable.ListMap
+import scala.util.{Failure, Success, Try}
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat, CatalogTable, HiveTableRelation}
+import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, AttributeSet, NamedExpression}
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.connector.catalog.{CatalogPlugin, Identifier, TableCatalog}
+import org.apache.spark.sql.execution.datasources.LogicalRelation
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2ScanRelation
+
+import org.apache.kyuubi.Logging
+import org.apache.kyuubi.plugin.lineage.events.Lineage
+import org.apache.kyuubi.plugin.lineage.helper.SparkListenerHelper.isSparkVersionAtMost
+
+trait LineageParser {
+ def sparkSession: SparkSession
+
+ type AttributeMap[A] = ListMap[Attribute, A]
+
+ def parse(plan: LogicalPlan): Lineage = {
+ val columnsLineage =
+ extractColumnsLineage(plan, ListMap[Attribute, AttributeSet]()).toList.collect {
+ case (k, attrs) =>
+ k.name -> attrs.map(_.name).toSet
+ }
+ val (inputTables, outputTables) = columnsLineage.foldLeft((List[String](), List[String]())) {
+ case ((inputs, outputs), (out, in)) =>
+ val x = (inputs ++ in.map(_.split('.').init.mkString("."))).filter(_.nonEmpty)
+ val y = outputs ++ List(out.split('.').init.mkString(".")).filter(_.nonEmpty)
+ (x, y)
+ }
+ Lineage(inputTables.distinct, outputTables.distinct, columnsLineage)
+ }
+
+ private def mergeColumnsLineage(
+ left: AttributeMap[AttributeSet],
+ right: AttributeMap[AttributeSet]): AttributeMap[AttributeSet] = {
+ left ++ right.map {
+ case (k, attrs) =>
+ k -> (attrs ++ left.getOrElse(k, AttributeSet.empty))
+ }
+ }
+
+ private def joinColumnsLineage(
+ parent: AttributeMap[AttributeSet],
+ child: AttributeMap[AttributeSet]): AttributeMap[AttributeSet] = {
+ if (parent.isEmpty) child
+ else {
+ val childMap = child.map { case (k, attrs) => (k.exprId, attrs) }
+ parent.map { case (k, attrs) =>
+ k -> AttributeSet(attrs.flatMap(attr =>
+ childMap.getOrElse(attr.exprId, AttributeSet.empty)))
+ }
+ }
+ }
+
+ private def getSelectColumnLineage(
+ named: Seq[NamedExpression]): AttributeMap[AttributeSet] = {
+ val exps = named.map {
+ case a: Alias => a.toAttribute -> a.references
+ case a: Attribute => a -> a.references
+ }
+ ListMap(exps: _*)
+ }
+
+ private def joinRelationColumnLineage(
+ parent: AttributeMap[AttributeSet],
+ relationAttrs: Seq[Attribute],
+ tableName: String = ""): AttributeMap[AttributeSet] = {
+ val relationAttrSet = AttributeSet(relationAttrs)
+ if (parent.nonEmpty) {
+ parent.map { case (k, attrs) =>
+ k -> AttributeSet(attrs.collect {
+ case attr if relationAttrSet.contains(attr) =>
+ attr.withName(Seq(tableName, attr.name).filter(_.nonEmpty).mkString("."))
+ })
+ }
+ } else {
+ ListMap(relationAttrs.map { attr =>
+ (
+ attr,
+ AttributeSet(attr.withName(Seq(tableName, attr.name).filter(_.nonEmpty).mkString("."))))
+ }: _*)
+ }
+ }
+
+ private def extractColumnsLineage(
+ plan: LogicalPlan,
+ parentColumnsLineage: AttributeMap[AttributeSet]): AttributeMap[AttributeSet] = {
+
+ def getPlanField[T](field: String): T = {
+ getFieldVal[T](plan, field)
+ }
+
+ def getCurrentPlanField[T](curPlan: LogicalPlan, field: String): T = {
+ getFieldVal[T](curPlan, field)
+ }
+
+ def getPlanMethod[T](name: String): T = {
+ getMethod[T](plan, name)
+ }
+
+ def getQuery: LogicalPlan = {
+ getPlanField[LogicalPlan]("query")
+ }
Review Comment:
better to move these methods out since we are inside recursion
##########
extensions/spark/kyuubi-spark-listener/src/main/scala/org/apache/kyuubi/plugin/lineage/events/OperationLineageEvent.scala:
##########
@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.lineage.events
+
+import org.apache.kyuubi.Utils
+import org.apache.kyuubi.events.KyuubiEvent
+
+case class Lineage(
Review Comment:
can we add some comments to list some examples of the `Lineage`, it may be used by developers.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] pan3793 commented on a diff in pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
pan3793 commented on code in PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#discussion_r940895465
##########
externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/events/SparkOperationEvent.scala:
##########
@@ -29,23 +29,23 @@ import org.apache.kyuubi.events.KyuubiEvent
/**
* A [[SparkOperationEvent]] used to tracker the lifecycle of an operation at Spark SQL Engine side.
* <ul>
- * <li>Operation Basis</li>
- * <li>Operation Live Status</li>
- * <li>Parent Session Id</li>
+ * <li>Operation Basis</li>
+ * <li>Operation Live Status</li>
+ * <li>Parent Session Id</li>
* </ul>
*
- * @param statementId the unique identifier of a single operation
- * @param statement the sql that you execute
+ * @param statementId the unique identifier of a single operation
+ * @param statement the sql that you execute
* @param shouldRunAsync the flag indicating whether the query runs synchronously or not
- * @param state the current operation state
- * @param eventTime the time when the event created & logged
- * @param createTime the time for changing to the current operation state
- * @param startTime the time the query start to time of this operation
- * @param completeTime time time the query ends
- * @param exception: caught exception if have
- * @param sessionId the identifier of the parent session
- * @param sessionUser the authenticated client user
- * @param executionId the query execution id of this operation
+ * @param state the current operation state
+ * @param eventTime the time when the event created & logged
+ * @param createTime the time for changing to the current operation state
+ * @param startTime the time the query start to time of this operation
+ * @param completeTime time time the query ends
+ * @param exception : caught exception if have
Review Comment:
nit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] pan3793 commented on a diff in pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
pan3793 commented on code in PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#discussion_r952423846
##########
extensions/spark/kyuubi-spark-lineage/pom.xml:
##########
@@ -0,0 +1,103 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ Licensed to the Apache Software Foundation (ASF) under one or more
+ ~ contributor license agreements. See the NOTICE file distributed with
+ ~ this work for additional information regarding copyright ownership.
+ ~ The ASF licenses this file to You under the Apache License, Version 2.0
+ ~ (the "License"); you may not use this file except in compliance with
+ ~ the License. You may obtain a copy of the License at
+ ~
+ ~ http://www.apache.org/licenses/LICENSE-2.0
+ ~
+ ~ Unless required by applicable law or agreed to in writing, software
+ ~ distributed under the License is distributed on an "AS IS" BASIS,
+ ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ ~ See the License for the specific language governing permissions and
+ ~ limitations under the License.
+ -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <parent>
+ <artifactId>kyuubi-parent</artifactId>
+ <groupId>org.apache.kyuubi</groupId>
+ <version>1.7.0-SNAPSHOT</version>
+ <relativePath>../../../pom.xml</relativePath>
+
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+
+ <artifactId>kyuubi-spark-lineage_2.12</artifactId>
+ <name>Kyuubi Dev Spark Lineage Extension</name>
+ <packaging>jar</packaging>
+ <url>https://kyuubi.apache.org/</url>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.kyuubi</groupId>
+ <artifactId>kyuubi-common_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.kyuubi</groupId>
+ <artifactId>kyuubi-events_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.spark</groupId>
+ <artifactId>spark-sql_${scala.binary.version}</artifactId>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>commons-collections</groupId>
+ <artifactId>commons-collections</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.kyuubi</groupId>
+ <artifactId>kyuubi-common_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.spark</groupId>
+ <artifactId>spark-catalyst_${scala.binary.version}</artifactId>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.spark</groupId>
+ <artifactId>spark-sql_${scala.binary.version}</artifactId>
+ <version>${spark.version}</version>
Review Comment:
remove version, it should be defined at root pom
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] cfmcgrady commented on a diff in pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
cfmcgrady commented on code in PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#discussion_r952319309
##########
extensions/spark/kyuubi-spark-lineage/src/test/scala/org/apache/spark/sql/SparkListenerExtenstionTest.scala:
##########
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql
+
+import java.nio.file.Files
+
+import org.apache.spark.SparkConf
+
+trait SparkListenerExtensionTest {
+ protected val catalogImpl: String
+ protected def format: String = if (catalogImpl == "hive") "hive" else "parquet"
+
+ protected lazy val spark: SparkSession = {
+ val metastore = {
+ val path = Files.createTempDirectory("hms")
Review Comment:
nit: create a temporary directory via `org.apache.spark.util.Utils.createTempDir()`, like we did in the `KyuubiSparkSQLExtensionTest` class, then the directory will be automatically deleted by the shutdown hook.
https://github.com/apache/incubator-kyuubi/blob/1642de3f1a2b303e6c74002a995912038fc1f860/extensions/spark/kyuubi-extension-spark-common/src/test/scala/org/apache/spark/sql/KyuubiSparkSQLExtensionTest.scala#L78-L93
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] iodone commented on a diff in pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
iodone commented on code in PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#discussion_r952488737
##########
extensions/spark/kyuubi-spark-lineage/src/main/scala/org/apache/kyuubi/plugin/lineage/events/OperationLineageEvent.scala:
##########
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.lineage.events
+
+import org.apache.kyuubi.Utils
+import org.apache.kyuubi.events.KyuubiEvent
+
+case class ColumnLineage(column: String, ordinal: Int, originalColumns: Set[String])
+
+/**
+ * @param inputTables the tables of the operation will read
+ * @param outputTables the tables of the operation will write
+ * @param columnLineage the output columns are associated to columns of the real table's columns
+ */
+
+class Lineage(
+ val inputTables: List[String],
+ val outputTables: List[String],
+ val columnLineage: List[ColumnLineage]) {
+
+ override def equals(other: Any): Boolean = other match {
Review Comment:
this for compatibility with test code
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] ulysses-you commented on pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
ulysses-you commented on PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#issuecomment-1217381579
It seems Kyuubi can provide a way for Spark extensions to inject event to Kyuubi event, so we can always unify the Spark event into Kyuubi event. Maybe something like user can get KyuubiSession by `KyuubiSession.get` in Spark extension. What do you think ? @iodone @yaooqinn @pan3793 @turboFei
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] bowenliang123 commented on pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
bowenliang123 commented on PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#issuecomment-1212900196
Will keep following this feature. I am looking for an Atlas hook as lineage plugin for Spark sql or Kyuubi .
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] iodone commented on a diff in pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
iodone commented on code in PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#discussion_r952123129
##########
docs/extensions/engines/spark/lineage.md:
##########
@@ -0,0 +1,171 @@
+<!--
+ - Licensed to the Apache Software Foundation (ASF) under one or more
+ - contributor license agreements. See the NOTICE file distributed with
+ - this work for additional information regarding copyright ownership.
+ - The ASF licenses this file to You under the Apache License, Version 2.0
+ - (the "License"); you may not use this file except in compliance with
+ - the License. You may obtain a copy of the License at
+ -
+ - http://www.apache.org/licenses/LICENSE-2.0
+ -
+ - Unless required by applicable law or agreed to in writing, software
+ - distributed under the License is distributed on an "AS IS" BASIS,
+ - WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ - See the License for the specific language governing permissions and
+ - limitations under the License.
+ -->
+
+# SQL Lineage Support
+
+When we execute a SQL statement, the computing engine will return to a result set,
+each column in the result set may originate from different tables, and these different
+tables depend on other tables, which may have been calculated by functions, aggregation, etc.
+The source table is related to the result set, which is the SQL Lineage.
+
+## Introduction
+
+The current lineage parsing functionality is implemented as a plugin by extending Spark's `QueryExecutionListener`.
+1. The `SparkListenerSQLExecutionEnd` event is triggered after the SQL execution is finished and captured by the `QueryExecuctionListener`,
+ where the SQL lineage parsing process is performed on the successfully executed SQL.
+2. Will write the parsed lineage information to the log file in JSON format.
+
+### Example
+
+When the following SQL is executed:
+```sql
+select a as col0, b as col1 from test_table0
+```
+The lineage of this SQL:
+```json
+{
+ "inputTables":[
+ "default.test_table0"
+ ],
+ "outputTables":[
+
+ ],
+ "columnLineage":[
+ [
+ "col0",
+ [
+ "default.test_table0.a"
+ ]
+ ],
+ [
+ "col1",
+ [
+ "default.test_table0.b"
+ ]
+ ]
+ ]
Review Comment:
yes，i will improve it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] cfmcgrady commented on a diff in pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
cfmcgrady commented on code in PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#discussion_r952319309
##########
extensions/spark/kyuubi-spark-lineage/src/test/scala/org/apache/spark/sql/SparkListenerExtenstionTest.scala:
##########
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql
+
+import java.nio.file.Files
+
+import org.apache.spark.SparkConf
+
+trait SparkListenerExtensionTest {
+ protected val catalogImpl: String
+ protected def format: String = if (catalogImpl == "hive") "hive" else "parquet"
+
+ protected lazy val spark: SparkSession = {
+ val metastore = {
+ val path = Files.createTempDirectory("hms")
Review Comment:
nit: create a temporary directory via `org.apache.spark.util.Utils.Utils.createTempDir()`, like we did in the `KyuubiSparkSQLExtensionTest` class, then the directory will be automatically deleted by the shutdown hook.
https://github.com/apache/incubator-kyuubi/blob/1642de3f1a2b303e6c74002a995912038fc1f860/extensions/spark/kyuubi-extension-spark-common/src/test/scala/org/apache/spark/sql/KyuubiSparkSQLExtensionTest.scala#L78-L93
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] bowenliang123 commented on pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
bowenliang123 commented on PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#issuecomment-1212900052
Will keep following this feature. I am looking for an Atlas hook as lineage plugin for Spark sql or Kyuubi .
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] yaooqinn commented on pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
yaooqinn commented on PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#issuecomment-1213003738
and maybe the lineage info shall have its own output logic
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] iodone commented on pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
iodone commented on PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#issuecomment-1216529596
Two approaches are currently implemented：
1. Spark SQL Engine , `set kyuubi.operation.lineage.enabled = true` to enable lineage information logging. The lineage information will be added to `SparkOperationEvent` and output to operation event json logger together.
```json
{
"Event": "org.apache.kyuubi.engine.spark.events.SparkOperationEvent",
"statementId": "ea7b2d7a-1301-4f82-aa79-797ccab43d4a",
"statement": "select a as col0, b as col1 from test_table0;",
"shouldRunAsync": false,
"state": "FINISHED_STATE",
"eventTime": 1660650146416,
"createTime": 1660650145261,
"startTime": 1660650145311,
"completeTime": 1660650146415,
"exception": null,
"sessionId": "e8f06b0f-5c5c-40ef-90de-c77fb4432d93",
"sessionUser": "work",
"executionId": 3,
"lineage": {
"inputTables": ["default.test_table0"],
"outputTables": [],
"columnLineage": [
["col0", ["default.test_table0.a"]],
["col1", ["default.test_table0.b"]]
]
},
"eventType": "spark_operation"
```
2. Spark Listener Extension Plugging, which is not strongly bound to Kyuubi Engine, can be tried to raw Spark. It has its own independent lineage information output, also output in json logger.
```json
{
"executionId": 2,
"eventTime": 1660650415849,
"lineage": {
"inputTables": ["default.test_table0"],
"outputTables": [],
"columnLineage": [
["col0", ["default.test_table0.a"]],
["col1", ["default.test_table0.b"]]
]
},
"exception": null,
"eventType": "operation_lineage"
}
```
In way 1, the lineage information can be combined with sql event information to easy to collection and statistics.
In way 2, sql operation event logger and lineage event logger are independent of each other and can only be associated by `executionId`. After collection, we need to do the correlation process.
Do we need to keep all these two ways? @yaooqinn @ulysses-you
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] iodone commented on a diff in pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
iodone commented on code in PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#discussion_r951084988
##########
docs/extensions/engines/spark/lineage.md:
##########
@@ -0,0 +1,168 @@
+<!--
+ - Licensed to the Apache Software Foundation (ASF) under one or more
+ - contributor license agreements. See the NOTICE file distributed with
+ - this work for additional information regarding copyright ownership.
+ - The ASF licenses this file to You under the Apache License, Version 2.0
+ - (the "License"); you may not use this file except in compliance with
+ - the License. You may obtain a copy of the License at
+ -
+ - http://www.apache.org/licenses/LICENSE-2.0
+ -
+ - Unless required by applicable law or agreed to in writing, software
+ - distributed under the License is distributed on an "AS IS" BASIS,
+ - WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ - See the License for the specific language governing permissions and
+ - limitations under the License.
+ -->
+
+# SQL Lineage Support
+
+When we execute a SQL statement, the computing engine will return to a result set,
+each column in the result set may originate from different tables, and these different
+tables depend on other tables, which may have been calculated by functions, aggregation, etc.
+The source table is related to the result set, which is the SQL Lineage.
+
+## Introduction
+
+The current lineage parsing functionality is implemented as a plugin by extending Spark's `QueryExecutionListener`.
+1. The `SparkListenerSQLExecutionEnd` event is triggered after the SQL execution is finished and captured by the `QueryExecuctionListener`,
+ where the SQL lineage parsing process is performed on the successfully executed SQL.
+2. Will write the parsed lineage information to the log file in JSON format.
+
+### Example
+
+When the following SQL is executed:
+```sql
+select a as col0, b as col1 from test_table0
+```
+The lineage of this SQL:
+```json
+{
+ "inputTables":[
+ "default.test_table0"
+ ],
+ "outputTables":[
+
+ ],
+ "columnLineage":[
+ [
Review Comment:
`List[(String, List[String])]` type be used here，index(0) refers to the column name，index(1) refers to the real column name, no encapsulation with `case class`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] pan3793 commented on a diff in pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
pan3793 commented on code in PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#discussion_r940895078
##########
externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/KyuubiSparkUtil.scala:
##########
@@ -90,4 +90,59 @@ object KyuubiSparkUtil extends Logging {
val runtimeSparkVer = org.apache.spark.SPARK_VERSION
Utils.majorMinorVersion(runtimeSparkVer)
}
+
+ /**
+ * Given a Kyuubi/Spark/Hive version string,
+ * return the (major version number, minor version number).
+ * E.g., for 2.0.1-SNAPSHOT, return (2, 0).
+ */
+ def majorMinorVersion(version: String): (Int, Int) = {
Review Comment:
It can be simplified since https://github.com/apache/incubator-kyuubi/pull/3181 get merged.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] iodone commented on pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
iodone commented on PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#issuecomment-1212729657
@ulysses-you I refactored some code，could you help me to review it?
Thanks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] yaooqinn commented on a diff in pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
yaooqinn commented on code in PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#discussion_r947400952
##########
extensions/spark/kyuubi-spark-listener/src/main/scala/org/apache/kyuubi/listener/SparkOperationLineageQueryExecutionListener.scala:
##########
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.listener
Review Comment:
org.apache.kyuubi.plugin.lineage
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] iodone commented on pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
iodone commented on PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#issuecomment-1217417774
> @ulysses-you can't the current event framework achieve such a goal? with custom event and costume event handler?
```scala
class SparkHistoryLoggingEventHandler(sc: SparkContext) extends EventHandler[KyuubiEvent] {
override def apply(event: KyuubiEvent): Unit = {
SparkContextHelper.postEventToSparkListenerBus(event, sc)
}
}
```
Here's the corresponding implementation that writes a custom `SparkOperationKyuubiEvent` to the `SparkListenerBus`, which I think can catch the event by extends spark extra listener.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] iodone commented on a diff in pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
iodone commented on code in PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#discussion_r944202694
##########
externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/spark/kyuubi/SparkSQLLineageParseHelper.scala:
##########
@@ -0,0 +1,309 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.kyuubi
+
+import scala.collection.immutable.ListMap
+import scala.util.{Failure, Success, Try}
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat, CatalogTable, HiveTableRelation}
+import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, AttributeSet, NamedExpression}
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.connector.catalog.{CatalogPlugin, Identifier, TableCatalog}
+import org.apache.spark.sql.execution.datasources.LogicalRelation
+import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, DataSourceV2ScanRelation}
+
+import org.apache.kyuubi.Logging
+import org.apache.kyuubi.engine.spark.KyuubiSparkUtil.isSparkVersionAtMost
+import org.apache.kyuubi.engine.spark.events.Lineage
+
+trait LineageParser {
+ def sparkSession: SparkSession
+
+ type AttributeMap[A] = ListMap[Attribute, A]
+
+ def parse(plan: LogicalPlan): Lineage = {
+ val columnsLineage =
+ extractColumnsLineage(plan, ListMap[Attribute, AttributeSet]()).toList.collect {
+ case (k, attrs) =>
+ k.name -> attrs.map(_.name).toSet
+ }
+ val (inputTables, outputTables) = columnsLineage.foldLeft((List[String](), List[String]())) {
+ case ((inputs, outputs), (out, in)) =>
+ val x = (inputs ++ in.map(_.split('.').init.mkString("."))).filter(_.nonEmpty)
+ val y = outputs ++ List(out.split('.').init.mkString(".")).filter(_.nonEmpty)
+ (x, y)
+ }
+ Lineage(inputTables.distinct, outputTables.distinct, columnsLineage)
+ }
+
+ private def mergeColumnsLineage(
+ left: AttributeMap[AttributeSet],
+ right: AttributeMap[AttributeSet]): AttributeMap[AttributeSet] = {
+ left ++ right.map {
+ case (k, attrs) =>
+ k -> (attrs ++ left.getOrElse(k, AttributeSet.empty))
+ }
+ }
+
+ private def joinColumnsLineage(
+ parent: AttributeMap[AttributeSet],
+ child: AttributeMap[AttributeSet]): AttributeMap[AttributeSet] = {
+ if (parent.isEmpty) child
+ else {
+ val childMap = child.map { case (k, attrs) => (k.exprId, attrs) }
+ parent.map { case (k, attrs) =>
+ k -> AttributeSet(attrs.flatMap(attr =>
+ childMap.getOrElse(attr.exprId, AttributeSet.empty)))
+ }
+ }
+ }
+
+ private def getSelectColumnLineage(
+ named: Seq[NamedExpression]): AttributeMap[AttributeSet] = {
+ val exps = named.map {
+ case a: Alias => a.toAttribute -> a.references
+ case a: Attribute => a -> a.references
+ }
+ ListMap(exps: _*)
+ }
+
+ private def joinRelationColumnLineage(
+ parent: AttributeMap[AttributeSet],
+ relationAttrs: Seq[Attribute],
+ tableName: String = ""): AttributeMap[AttributeSet] = {
+ val relationAttrSet = AttributeSet(relationAttrs)
+ if (parent.nonEmpty) {
+ parent.map { case (k, attrs) =>
+ k -> AttributeSet(attrs.collect {
+ case attr if relationAttrSet.contains(attr) =>
+ attr.withName(Seq(tableName, attr.name).filter(_.nonEmpty).mkString("."))
+ })
+ }
+ } else {
+ ListMap(relationAttrs.map { attr =>
+ (
+ attr,
+ AttributeSet(attr.withName(Seq(tableName, attr.name).filter(_.nonEmpty).mkString("."))))
+ }: _*)
+ }
+ }
+
+ private def extractColumnsLineage(
+ plan: LogicalPlan,
+ parentColumnsLineage: AttributeMap[AttributeSet]): AttributeMap[AttributeSet] = {
+
+ def getPlanField[T](field: String): T = {
+ getFieldVal[T](plan, field)
+ }
+
+ def getCurrentPlanField[T](curPlan: LogicalPlan, field: String): T = {
+ getFieldVal[T](curPlan, field)
+ }
+
+ def getPlanMethod[T](name: String): T = {
+ getMethod[T](plan, name)
+ }
+
+ def getQuery: LogicalPlan = {
+ getPlanField[LogicalPlan]("query")
+ }
+
+ plan match {
+ // For command
+ case p if p.nodeName == "AlterViewAsCommand" =>
+ val analyzed = sparkSession.sessionState.analyzer.execute(getQuery)
+ val view = getPlanField[TableIdentifier]("name").unquotedString
+ extractColumnsLineage(analyzed, parentColumnsLineage).map { case (k, v) =>
+ k.withName(s"$view.${k.name}") -> v
+ }
+
+ case p if p.nodeName == "CreateViewCommand" =>
+ val view = getPlanField[TableIdentifier]("name").unquotedString
+ val outputCols =
+ getPlanField[Seq[(String, Option[String])]]("userSpecifiedColumns").map(_._1)
+ val query =
+ if (isSparkVersionAtMost("3.1")) {
+ getPlanField[LogicalPlan]("child")
+ } else {
+ getPlanField[LogicalPlan]("plan")
+ }
+ val analyzed = sparkSession.sessionState.analyzer.execute(query)
+ extractColumnsLineage(analyzed, parentColumnsLineage).zipWithIndex.map {
+ case ((k, v), i) if outputCols.nonEmpty => k.withName(s"$view.${outputCols(i)}") -> v
+ case ((k, v), _) => k.withName(s"$view.${k.name}") -> v
+ }
+
+ case p if p.nodeName == "CreateDataSourceTableAsSelectCommand" =>
+ val table = getPlanField[CatalogTable]("table").qualifiedName
+ val analyzed = sparkSession.sessionState.analyzer.execute(getQuery)
+ extractColumnsLineage(analyzed, parentColumnsLineage).map { case (k, v) =>
+ k.withName(s"$table.${k.name}") -> v
+ }
+
+ case p
+ if p.nodeName == "CreateHiveTableAsSelectCommand" ||
+ p.nodeName == "OptimizedCreateHiveTableAsSelectCommand" =>
+ val table = getPlanField[CatalogTable]("tableDesc").qualifiedName
+ extractColumnsLineage(getQuery, parentColumnsLineage).map { case (k, v) =>
+ k.withName(s"$table.${k.name}") -> v
+ }
+
+ case p
+ if p.nodeName == "CreateTableAsSelect" ||
+ p.nodeName == "ReplaceTableAsSelect" =>
+ val (table, namespace, catalog) =
+ if (isSparkVersionAtMost("3.2")) {
+ (
+ getPlanField[Identifier]("tableName").name,
+ getPlanField[Identifier]("tableName").namespace.mkString("."),
+ getPlanField[TableCatalog]("catalog").name())
+ } else {
+ (
+ getPlanMethod[Identifier]("tableName").name(),
+ getPlanMethod[Identifier]("tableName").namespace().mkString("."),
+ getCurrentPlanField[CatalogPlugin](
+ getPlanMethod[LogicalPlan]("left"),
+ "catalog").name())
+ }
+ extractColumnsLineage(getQuery, parentColumnsLineage).map { case (k, v) =>
+ k.withName(s"$catalog.$namespace.$table.${k.name}") -> v
+ }
+
+ case p if p.nodeName == "InsertIntoDataSourceCommand" =>
+ val logicalRelation = getPlanField[LogicalRelation]("logicalRelation")
+ val table = logicalRelation.catalogTable.map(_.qualifiedName).getOrElse("")
+ extractColumnsLineage(getQuery, parentColumnsLineage).map { case (k, v) =>
+ k.withName(s"$table.${k.name}") -> v
+ }
+
+ case p if p.nodeName == "InsertIntoHadoopFsRelationCommand" =>
+ val table =
+ getPlanField[Option[CatalogTable]]("catalogTable").map(_.qualifiedName).getOrElse("")
+ extractColumnsLineage(getQuery, parentColumnsLineage).map { case (k, v) =>
+ k.withName(s"$table.${k.name}") -> v
+ }
+
+ case p
+ if p.nodeName == "InsertIntoDataSourceDirCommand" ||
+ p.nodeName == "InsertIntoHiveDirCommand" =>
+ val dir =
+ getPlanField[CatalogStorageFormat]("storage").locationUri.map(_.toString).getOrElse("")
+ extractColumnsLineage(getQuery, parentColumnsLineage).map { case (k, v) =>
+ k.withName(s"`$dir`.${k.name}") -> v
+ }
+
+ case p if p.nodeName == "InsertIntoHiveTable" =>
+ val table = getPlanField[CatalogTable]("table").qualifiedName
+ extractColumnsLineage(getQuery, parentColumnsLineage).map { case (k, v) =>
+ k.withName(s"$table.${k.name}") -> v
+ }
+
+ case p if p.nodeName == "SaveIntoDataSourceCommand" =>
+ extractColumnsLineage(getQuery, parentColumnsLineage)
+
+ // For query
+ case p: Project =>
+ val nextColumnsLineage =
+ joinColumnsLineage(parentColumnsLineage, getSelectColumnLineage(p.projectList))
+ p.children.map(extractColumnsLineage(_, nextColumnsLineage)).reduce(mergeColumnsLineage)
+
+ case p: Aggregate =>
+ val nextColumnsLineage =
+ joinColumnsLineage(parentColumnsLineage, getSelectColumnLineage(p.aggregateExpressions))
+ p.children.map(extractColumnsLineage(_, nextColumnsLineage)).reduce(mergeColumnsLineage)
+
+ case p: Union =>
+ // merge all children in to one derivedColumns
+ val childrenUnion =
+ p.children.map(extractColumnsLineage(_, ListMap[Attribute, AttributeSet]())).map(
+ _.values).reduce {
+ (left, right) =>
+ left.zip(right).map(attr => attr._1 ++ attr._2)
+ }
+ val childrenColumnsLineage = ListMap(p.output.zip(childrenUnion): _*)
+ joinColumnsLineage(parentColumnsLineage, childrenColumnsLineage)
+
+ case p: LogicalRelation if p.catalogTable.nonEmpty =>
+ val tableName = p.catalogTable.get.qualifiedName
+ joinRelationColumnLineage(parentColumnsLineage, p.output, tableName)
+
+ case p: HiveTableRelation =>
+ val tableName = p.tableMeta.qualifiedName
+ joinRelationColumnLineage(parentColumnsLineage, p.output, tableName)
+
+ case p: DataSourceV2Relation =>
+ val tableName = p.name
+ joinRelationColumnLineage(parentColumnsLineage, p.output, tableName)
+
+ case p: DataSourceV2ScanRelation =>
+ val tableName = p.name
+ joinRelationColumnLineage(parentColumnsLineage, p.output, tableName)
+
+ case p: LocalRelation =>
+ joinRelationColumnLineage(parentColumnsLineage, p.output)
Review Comment:
Nice
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] yaooqinn commented on pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
yaooqinn commented on PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#issuecomment-1212845121
I suggest this feature goes to extensions.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] ulysses-you commented on a diff in pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
ulysses-you commented on code in PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#discussion_r952092976
##########
docs/extensions/engines/spark/lineage.md:
##########
@@ -0,0 +1,171 @@
+<!--
+ - Licensed to the Apache Software Foundation (ASF) under one or more
+ - contributor license agreements. See the NOTICE file distributed with
+ - this work for additional information regarding copyright ownership.
+ - The ASF licenses this file to You under the Apache License, Version 2.0
+ - (the "License"); you may not use this file except in compliance with
+ - the License. You may obtain a copy of the License at
+ -
+ - http://www.apache.org/licenses/LICENSE-2.0
+ -
+ - Unless required by applicable law or agreed to in writing, software
+ - distributed under the License is distributed on an "AS IS" BASIS,
+ - WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ - See the License for the specific language governing permissions and
+ - limitations under the License.
+ -->
+
+# SQL Lineage Support
+
+When we execute a SQL statement, the computing engine will return to a result set,
+each column in the result set may originate from different tables, and these different
+tables depend on other tables, which may have been calculated by functions, aggregation, etc.
+The source table is related to the result set, which is the SQL Lineage.
+
+## Introduction
+
+The current lineage parsing functionality is implemented as a plugin by extending Spark's `QueryExecutionListener`.
+1. The `SparkListenerSQLExecutionEnd` event is triggered after the SQL execution is finished and captured by the `QueryExecuctionListener`,
+ where the SQL lineage parsing process is performed on the successfully executed SQL.
+2. Will write the parsed lineage information to the log file in JSON format.
Review Comment:
depend on the spark engine event logger config, `kyuubi.engine.event.loggers`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] iodone commented on a diff in pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
iodone commented on code in PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#discussion_r951198951
##########
extensions/spark/kyuubi-spark-listener/src/main/scala/org/apache/kyuubi/plugin/lineage/events/OperationLineageEvent.scala:
##########
@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.lineage.events
+
+import org.apache.kyuubi.Utils
+import org.apache.kyuubi.events.KyuubiEvent
+
+case class Lineage(
Review Comment:
There will also be detailed instructions in the [docs/extensions/engines/spark/lineage.md](https://github.com/apache/incubator-kyuubi/pull/3185/files/e1abba76741e31bf66897532da43530516618173#diff-815ac878f18bdecd045087459e56c3d38eed6a967f8a2d300bb43aefd9e8a679)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] pan3793 commented on a diff in pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
pan3793 commented on code in PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#discussion_r951035041
##########
externals/kyuubi-spark-sql-engine/src/test/scala/org/apache/kyuubi/engine/spark/events/handler/SparkJsonLoggingEventHandlerSuite.scala:
##########
@@ -128,4 +138,30 @@ class SparkJsonLoggingEventHandlerSuite extends WithSparkSQLEngine with HiveJDBC
}
}
}
+
+ test("operation lineage event loggin: for execute sql") {
+ val statementEventPath = Paths.get(
+ logRoot,
+ "operation_lineage",
+ s"day=$currentDate",
+ spark.sparkContext.applicationId + ".json")
+ val table = statementEventPath.getParent
+
+ withTable("test_table0") { _ =>
+ spark.sql("create table test_table0(a string, b string)")
+ spark.sql("select a as col0, b as col1 from test_table0").collect()
+
+ val result = spark.sql(s"select * from `json`.`$table`")
+ val expected =
+ "WrappedArray(WrappedArray(col0, [\"default.test_table0.a\"]), " +
+ "WrappedArray(col1, [\"default.test_table0.b\"]))"
+
+ assert(result.select("lineage")
+ .collect().last
+ .getStruct(0)
+ .getAs("columnLineage")
+ .toString == expected)
Review Comment:
compare case class directly?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] iodone commented on pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
iodone commented on PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#issuecomment-1212958614
> I suggest this feature goes to extensions.
The original design was implemented in the extension layer, extending QueryExecutionListener, so we can take this approach, right?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] yaooqinn commented on pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
yaooqinn commented on PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#issuecomment-1217395959
Let's clean the way 1 first, I think 2 is better and can still be improved
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] ulysses-you commented on pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
ulysses-you commented on PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#issuecomment-1217404340
I mean we can reuse the functionality of Kyuubi engine event at extension, for example: `ENGINE_EVENT_LOGGERS`. So we do not need additional config and code to do the similar thing. To approach this, we can provide a global static instance like `KyuubiSparkSession` which provides some methods about injecting event, so that at Spark extension can use it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] yaooqinn commented on pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
yaooqinn commented on PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#issuecomment-1217415145
@ulysses-you can't the current event framework achieve such a goal? with custom event and costume event handler?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] iodone commented on a diff in pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
iodone commented on code in PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#discussion_r941084765
##########
externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/spark/kyuubi/SparkSQLLineageParseHelper.scala:
##########
@@ -0,0 +1,366 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.kyuubi
+
+import scala.collection.immutable.ListMap
+import scala.util.{Failure, Success, Try}
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat, CatalogTable, HiveTableRelation}
+import org.apache.spark.sql.catalyst.expressions.{Alias, NamedExpression}
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.connector.catalog.{CatalogPlugin, Identifier, TableCatalog}
+import org.apache.spark.sql.execution.datasources.LogicalRelation
+import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, DataSourceV2ScanRelation}
+
+import org.apache.kyuubi.Logging
+import org.apache.kyuubi.engine.spark.KyuubiSparkUtil.isSparkVersionAtMost
+import org.apache.kyuubi.engine.spark.events.Lineage
+
+trait LineageParser {
+ def sparkSession: SparkSession
+
+ def parse(plan: LogicalPlan): Lineage = {
+ val ret = {
+ extractColumnsLineage(plan, ListMap()).toList.collect {
+ case (k, columns) =>
+ k.split('#').head -> columns.map(_.split('#').head)
+ }
+ }
+ val (inputTables, outputTables) = ret.foldLeft((List[String](), List[String]())) {
+ case ((inputs, outputs), (out, in)) =>
+ val x = (inputs ++ in.map(_.split('.').init.mkString("."))).filter(_.nonEmpty)
+ val y = outputs ++ List(out.split('.').init.mkString(".")).filter(_.nonEmpty)
+ (x, y)
+ }
+ Lineage(inputTables.distinct, outputTables.distinct, ret)
+ }
+
+ private def mergeColumnsLineage(
+ left: ListMap[String, List[String]],
+ right: ListMap[String, List[String]]): ListMap[String, List[String]] = {
+ if (left.isEmpty || right.isEmpty) left ++ right
+ else {
+ val head =
+ ListMap(left.head._1 -> (left.head._2 ++ right.getOrElse(left.head._1, List.empty)))
+ head ++ mergeColumnsLineage(left.tail, right - left.head._1)
+ }
+ }
+
+ private def joinColumnsLineage(
+ parent: ListMap[String, List[String]],
+ child: ListMap[String, List[String]]): ListMap[String, List[String]] = {
+ if (parent.isEmpty) child
+ else {
+ val childTmp = child.map { case (k, v) => (k.split('#').last, v) }
+ parent.map { case (k, columns) =>
+ k -> {
+ columns.flatMap {
+ case col if childTmp.contains(col.split('#').last) => childTmp(col.split('#').last)
+ case _ => List()
+ }
+ }
+ }
+ }
+ }
+
+ private def extractColumnsLineage(
+ plan: LogicalPlan,
+ parentColumnsLineage: ListMap[String, List[String]]): ListMap[String, List[String]] = {
+
+ def getPlanField[T](field: String): T = {
+ getFieldVal[T](plan, field)
+ }
+
+ def getCurrentPlanField[T](curPlan: LogicalPlan, field: String): T = {
+ getFieldVal[T](curPlan, field)
+ }
+
+ def getPlanMethod[T](name: String): T = {
+ getMethod[T](plan, name)
+ }
+
+ def getQuery: LogicalPlan = {
+ getPlanField[LogicalPlan]("query")
+ }
+
+ plan match {
+ // For command
+ case p if p.nodeName == "AlterViewAsCommand" =>
+ val analyzed = sparkSession.sessionState.analyzer.execute(getQuery)
+ val view = getPlanField[TableIdentifier]("name").unquotedString
+ extractColumnsLineage(analyzed, parentColumnsLineage).map { case (k, v) =>
+ (s"$view.$k", v)
+ }
+
+ case p if p.nodeName == "CreateViewCommand" =>
+ val view = getPlanField[TableIdentifier]("name").unquotedString
+ val outputCols =
+ getPlanField[Seq[(String, Option[String])]]("userSpecifiedColumns").map(_._1)
+ val query =
+ if (isSparkVersionAtMost("3.1")) {
+ getPlanField[LogicalPlan]("child")
+ } else {
+ getPlanField[LogicalPlan]("plan")
+ }
+ val analyzed = sparkSession.sessionState.analyzer.execute(query)
+ extractColumnsLineage(analyzed, parentColumnsLineage).zipWithIndex.map {
+ case (x, i) if outputCols.nonEmpty => s"$view.${outputCols(i)}" -> x._2
+ case ((k, v), _) => s"$view.$k" -> v
+ }
+
+ case p if p.nodeName == "CreateDataSourceTableAsSelectCommand" =>
+ val table = getPlanField[CatalogTable]("table").qualifiedName
+ val analyzed = sparkSession.sessionState.analyzer.execute(getQuery)
+ extractColumnsLineage(analyzed, parentColumnsLineage).map { case (k, v) =>
+ s"$table.$k" -> v
+ }
+
+ case p
+ if p.nodeName == "CreateHiveTableAsSelectCommand" ||
+ p.nodeName == "OptimizedCreateHiveTableAsSelectCommand" =>
+ val table = getPlanField[CatalogTable]("tableDesc").qualifiedName
+ extractColumnsLineage(getQuery, parentColumnsLineage).map { case (k, v) =>
+ s"$table.$k" -> v
+ }
+
+ case p
+ if p.nodeName == "CreateTableAsSelect" ||
+ p.nodeName == "ReplaceTableAsSelect" =>
+ val (table, namespace, catalog) =
+ if (isSparkVersionAtMost("3.2")) {
+ (
+ getPlanField[Identifier]("tableName").name,
+ getPlanField[Identifier]("tableName").namespace.mkString("."),
+ getPlanField[TableCatalog]("catalog").name())
+ } else {
+ (
+ getPlanMethod[Identifier]("tableName").name(),
+ getPlanMethod[Identifier]("tableName").namespace().mkString("."),
+ getCurrentPlanField[CatalogPlugin](
+ getPlanMethod[LogicalPlan]("left"),
+ "catalog").name())
+ }
+ extractColumnsLineage(getQuery, parentColumnsLineage).map { case (k, v) =>
+ s"$catalog.$namespace.$table.$k" -> v
+ }
+
+ case p if p.nodeName == "InsertIntoDataSourceCommand" =>
+ val logicalRelation = getPlanField[LogicalRelation]("logicalRelation")
+ val table = logicalRelation.catalogTable.map(_.qualifiedName).getOrElse("")
+ extractColumnsLineage(getQuery, parentColumnsLineage).map { case (k, v) =>
+ s"$table.$k" -> v
+ }
+
+ case p if p.nodeName == "InsertIntoHadoopFsRelationCommand" =>
+ val table =
+ getPlanField[Option[CatalogTable]]("catalogTable").map(_.qualifiedName).getOrElse("")
+ extractColumnsLineage(getQuery, parentColumnsLineage).map { case (k, v) =>
+ s"$table.$k" -> v
+ }
+
+ case p
+ if p.nodeName == "InsertIntoDataSourceDirCommand" ||
+ p.nodeName == "InsertIntoHiveDirCommand" =>
+ val dir =
+ getPlanField[CatalogStorageFormat]("storage").locationUri.map(_.toString).getOrElse("")
+ extractColumnsLineage(getQuery, parentColumnsLineage).map { case (k, v) =>
+ s"`$dir`.$k" -> v
+ }
+
+ case p if p.nodeName == "InsertIntoHiveTable" =>
+ val table = getPlanField[CatalogTable]("table").qualifiedName
+ extractColumnsLineage(getQuery, parentColumnsLineage).map { case (k, v) =>
+ s"$table.$k" -> v
+ }
+
+ case p if p.nodeName == "SaveIntoDataSourceCommand" =>
+ // TODO: Should get the table via datasource options?
+ extractColumnsLineage(getQuery, parentColumnsLineage)
+
+ // For query
+ case p: Project =>
+ val nextColumnsLineage =
+ joinColumnsLineage(parentColumnsLineage, getSelectColumnLineage(p.projectList))
+ p.children.map(extractColumnsLineage(_, nextColumnsLineage)).reduce(mergeColumnsLineage)
+
+ case p: Aggregate =>
+ val nextColumnsLineage =
+ joinColumnsLineage(parentColumnsLineage, getSelectColumnLineage(p.aggregateExpressions))
+ p.children.map(extractColumnsLineage(_, nextColumnsLineage)).reduce(mergeColumnsLineage)
+
+ case p: Union =>
+ // merge all children in to one derivedColumns
+ val childrenUnion =
+ p.children.map(extractColumnsLineage(_, ListMap())).map(_.values).reduce {
+ (left, right) =>
+ left.zip(right).map(col => (col._1 ++ col._2).distinct)
+ }
+ val childrenColumnsLineage =
+ ListMap(p.output.map(a => s"${a.name}#${a.exprId.id}").toList.zip(childrenUnion): _*)
+ joinColumnsLineage(parentColumnsLineage, childrenColumnsLineage)
+
+ case p: SubqueryAlias =>
+ extractColumnsLineage(p.child, parentColumnsLineage)
Review Comment:
Yes, at first it was analyzed plan, but finally it was changed to optmized plan. we can remove it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] ulysses-you commented on a diff in pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
ulysses-you commented on code in PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#discussion_r952091074
##########
externals/kyuubi-spark-sql-engine/pom.xml:
##########
@@ -45,6 +45,13 @@
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.kyuubi</groupId>
+ <artifactId>kyuubi-spark-lineage_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
Review Comment:
In general, we do not make engine depend on any extension, even for test. If user want to enable one extension, just put the jar into engine classpath.
You can move the related test to lineage module.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] iodone commented on a diff in pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
iodone commented on code in PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#discussion_r952362012
##########
extensions/spark/kyuubi-spark-lineage/src/test/scala/org/apache/kyuubi/plugin/lineage/helper/SparkSQLLineageParserHelperSuite.scala:
##########
@@ -0,0 +1,722 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.lineage.helper
+
+import scala.collection.immutable.List
+import scala.reflect.io.File
+
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.{DataFrame, SparkListenerExtensionTest, SparkSession, SQLContext}
+import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat, CatalogTable, CatalogTableType}
+import org.apache.spark.sql.sources.{BaseRelation, InsertableRelation, SchemaRelationProvider}
+import org.apache.spark.sql.types.{IntegerType, StringType, StructType}
+
+import org.apache.kyuubi.KyuubiFunSuite
+import org.apache.kyuubi.plugin.lineage.events.Lineage
+import org.apache.kyuubi.plugin.lineage.helper.SparkListenerHelper.isSparkVersionAtMost
+
+class SparkSQLLineageParserHelperSuite extends KyuubiFunSuite
+ with SparkListenerExtensionTest {
+
+ val catalogName =
+ if (isSparkVersionAtMost("3.1")) "org.apache.spark.sql.connector.InMemoryTableCatalog"
+ else "org.apache.spark.sql.connector.catalog.InMemoryTableCatalog"
+
+ override protected val catalogImpl: String = "hive"
+
+ override def sparkConf(): SparkConf = {
+ super.sparkConf()
+ .set(
+ "spark.sql.catalog.v2_catalog",
+ catalogName)
+ }
+
+ override def beforeAll(): Unit = {
+ super.beforeAll()
+ spark.sql(s"create database if not exists test_db")
+ spark.sql(s"create database if not exists test_db0")
+ spark.sql(s"create table if not exists test_db0.test_table0" +
+ s" (key int, value string) using parquet")
+ spark.sql(s"create table if not exists test_db0.test_table_part0" +
+ s" (key int, value string, pid string) using parquet" +
Review Comment:
Some tables were created in the specific test, including both hive and parquet table
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] yaooqinn commented on pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
yaooqinn commented on PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#issuecomment-1213001452
> default.test_table0.a
yes, we can create a new module like the `kyuubi-spark-authz` module
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] ulysses-you commented on a diff in pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
ulysses-you commented on code in PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#discussion_r947422538
##########
extensions/spark/kyuubi-spark-listener/src/main/scala/org/apache/kyuubi/events/LineageEventHandlerRegister.scala:
##########
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.events
+
+import org.apache.spark.sql.SparkSession
+
+import org.apache.kyuubi.SparkListenerSQLConf
+
+object LineageEventHandlerRegister {
+
+ @volatile var isRegister: Boolean = false
+
+ def register(spark: SparkSession): Unit = {
+ if (!isRegister) {
+ val logPath =
+ spark.conf.getOption(SparkListenerSQLConf.LINEAGE_EVENT_JSON_LOG_PATH.key).getOrElse(
+ SparkListenerSQLConf.LINEAGE_EVENT_JSON_LOG_PATH.defaultValStr)
+ val handler = LineageJsonLoggingEventHandler(
+ spark.sparkContext.applicationAttemptId
+ .map(id => s"${spark.sparkContext.applicationId}_$id")
+ .getOrElse(spark.sparkContext.applicationId),
+ spark.sparkContext.hadoopConfiguration,
+ logPath)
+ EventBus.register[OperationLineageEvent](handler)
Review Comment:
do we need this ? we can decide how to handle event by `kyuubi.engine.event.loggers`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] ulysses-you commented on a diff in pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
ulysses-you commented on code in PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#discussion_r947396081
##########
externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/spark/kyuubi/SparkSQLLineageParseHelper.scala:
##########
@@ -0,0 +1,309 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.kyuubi
+
+import scala.collection.immutable.ListMap
+import scala.util.{Failure, Success, Try}
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat, CatalogTable, HiveTableRelation}
+import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, AttributeSet, NamedExpression}
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.connector.catalog.{CatalogPlugin, Identifier, TableCatalog}
+import org.apache.spark.sql.execution.datasources.LogicalRelation
+import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, DataSourceV2ScanRelation}
+
+import org.apache.kyuubi.Logging
+import org.apache.kyuubi.engine.spark.KyuubiSparkUtil.isSparkVersionAtMost
+import org.apache.kyuubi.engine.spark.events.Lineage
+
+trait LineageParser {
+ def sparkSession: SparkSession
+
+ type AttributeMap[A] = ListMap[Attribute, A]
+
+ def parse(plan: LogicalPlan): Lineage = {
+ val columnsLineage =
+ extractColumnsLineage(plan, ListMap[Attribute, AttributeSet]()).toList.collect {
+ case (k, attrs) =>
+ k.name -> attrs.map(_.name).toSet
+ }
+ val (inputTables, outputTables) = columnsLineage.foldLeft((List[String](), List[String]())) {
+ case ((inputs, outputs), (out, in)) =>
+ val x = (inputs ++ in.map(_.split('.').init.mkString("."))).filter(_.nonEmpty)
+ val y = outputs ++ List(out.split('.').init.mkString(".")).filter(_.nonEmpty)
+ (x, y)
+ }
+ Lineage(inputTables.distinct, outputTables.distinct, columnsLineage)
+ }
+
+ private def mergeColumnsLineage(
+ left: AttributeMap[AttributeSet],
+ right: AttributeMap[AttributeSet]): AttributeMap[AttributeSet] = {
+ left ++ right.map {
+ case (k, attrs) =>
+ k -> (attrs ++ left.getOrElse(k, AttributeSet.empty))
+ }
+ }
+
+ private def joinColumnsLineage(
+ parent: AttributeMap[AttributeSet],
+ child: AttributeMap[AttributeSet]): AttributeMap[AttributeSet] = {
+ if (parent.isEmpty) child
+ else {
+ val childMap = child.map { case (k, attrs) => (k.exprId, attrs) }
+ parent.map { case (k, attrs) =>
+ k -> AttributeSet(attrs.flatMap(attr =>
+ childMap.getOrElse(attr.exprId, AttributeSet.empty)))
+ }
+ }
+ }
+
+ private def getSelectColumnLineage(
+ named: Seq[NamedExpression]): AttributeMap[AttributeSet] = {
+ val exps = named.map {
+ case a: Alias => a.toAttribute -> a.references
+ case a: Attribute => a -> a.references
+ }
+ ListMap(exps: _*)
+ }
+
+ private def joinRelationColumnLineage(
+ parent: AttributeMap[AttributeSet],
+ relationAttrs: Seq[Attribute],
+ tableName: String = ""): AttributeMap[AttributeSet] = {
+ val relationAttrSet = AttributeSet(relationAttrs)
+ if (parent.nonEmpty) {
+ parent.map { case (k, attrs) =>
+ k -> AttributeSet(attrs.collect {
+ case attr if relationAttrSet.contains(attr) =>
+ attr.withName(Seq(tableName, attr.name).filter(_.nonEmpty).mkString("."))
+ })
+ }
+ } else {
+ ListMap(relationAttrs.map { attr =>
+ (
+ attr,
+ AttributeSet(attr.withName(Seq(tableName, attr.name).filter(_.nonEmpty).mkString("."))))
+ }: _*)
+ }
+ }
+
+ private def extractColumnsLineage(
+ plan: LogicalPlan,
+ parentColumnsLineage: AttributeMap[AttributeSet]): AttributeMap[AttributeSet] = {
+
+ def getPlanField[T](field: String): T = {
+ getFieldVal[T](plan, field)
+ }
+
+ def getCurrentPlanField[T](curPlan: LogicalPlan, field: String): T = {
+ getFieldVal[T](curPlan, field)
+ }
+
+ def getPlanMethod[T](name: String): T = {
+ getMethod[T](plan, name)
+ }
+
+ def getQuery: LogicalPlan = {
+ getPlanField[LogicalPlan]("query")
+ }
+
+ plan match {
+ // For command
+ case p if p.nodeName == "AlterViewAsCommand" =>
+ val analyzed = sparkSession.sessionState.analyzer.execute(getQuery)
Review Comment:
oh, I see
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] yaooqinn commented on pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
yaooqinn commented on PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#issuecomment-1217393500
I do not get your point @ulysses-you
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] ulysses-you commented on a diff in pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
ulysses-you commented on code in PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#discussion_r952091074
##########
externals/kyuubi-spark-sql-engine/pom.xml:
##########
@@ -45,6 +45,13 @@
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.kyuubi</groupId>
+ <artifactId>kyuubi-spark-lineage_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
Review Comment:
In general, we do not make engine depend on any extension. If user want to enable one extension, just put the jar into engine classpath.
You can move the related test to lineage module.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] ulysses-you commented on pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
ulysses-you commented on PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#issuecomment-1223444022
@iodone the style check job failed can be ignored since we do not install the new added module before style check
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] iodone commented on pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
iodone commented on PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#issuecomment-1223456729
> @iodone can you create a issue that support track lineage from subquery and v2write ? we can improve that in future
OK
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] pan3793 commented on a diff in pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
pan3793 commented on code in PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#discussion_r951032631
##########
docs/extensions/engines/spark/lineage.md:
##########
@@ -0,0 +1,168 @@
+<!--
+ - Licensed to the Apache Software Foundation (ASF) under one or more
+ - contributor license agreements. See the NOTICE file distributed with
+ - this work for additional information regarding copyright ownership.
+ - The ASF licenses this file to You under the Apache License, Version 2.0
+ - (the "License"); you may not use this file except in compliance with
+ - the License. You may obtain a copy of the License at
+ -
+ - http://www.apache.org/licenses/LICENSE-2.0
+ -
+ - Unless required by applicable law or agreed to in writing, software
+ - distributed under the License is distributed on an "AS IS" BASIS,
+ - WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ - See the License for the specific language governing permissions and
+ - limitations under the License.
+ -->
+
+# SQL Lineage Support
+
+When we execute a SQL statement, the computing engine will return to a result set,
+each column in the result set may originate from different tables, and these different
+tables depend on other tables, which may have been calculated by functions, aggregation, etc.
+The source table is related to the result set, which is the SQL Lineage.
+
+## Introduction
+
+The current lineage parsing functionality is implemented as a plugin by extending Spark's `QueryExecutionListener`.
+1. The `SparkListenerSQLExecutionEnd` event is triggered after the SQL execution is finished and captured by the `QueryExecuctionListener`,
+ where the SQL lineage parsing process is performed on the successfully executed SQL.
+2. Will write the parsed lineage information to the log file in JSON format.
+
+### Example
+
+When the following SQL is executed:
+```sql
+select a as col0, b as col1 from test_table0
+```
+The lineage of this SQL:
+```json
+{
+ "inputTables":[
+ "default.test_table0"
+ ],
+ "outputTables":[
+
+ ],
+ "columnLineage":[
+ [
Review Comment:
`{` ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] yaooqinn commented on pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
yaooqinn commented on PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#issuecomment-1217413620
> > Let's clean the way 1 first, I think 2 is better and can still be improved
>
> In some scenarios the user submits SQL only to get the lineage information and does not execute it. Just like `PlanOnlyMode`, we can output only the lineage information. In this case we have to implement the lineage extraction capability in Kyuubi Spark Engine. This implement can reuse the way 1
way 2 can be reused too and more clear
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] cxzl25 commented on a diff in pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
cxzl25 commented on code in PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#discussion_r947583860
##########
extensions/spark/kyuubi-spark-listener/src/main/scala/org/apache/kyuubi/events/LineageJsonLoggingEventHandler.scala:
##########
@@ -0,0 +1,104 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.events
+
+import java.io.{BufferedOutputStream, FileOutputStream, PrintWriter}
+import java.net.URI
+
+import scala.collection.mutable.HashMap
+
+import org.apache.commons.lang3.StringUtils
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.{FileSystem, FSDataOutputStream, Path}
+import org.apache.hadoop.fs.permission.FsPermission
+
+import org.apache.kyuubi.Logging
+import org.apache.kyuubi.events.handler.EventHandler
+
+case class LineageJsonLoggingEventHandler(
+ logName: String,
+ hadoopConf: Configuration,
+ logPath: String)
+ extends EventHandler[KyuubiEvent] with Logging {
+
+ import JsonLoggingEventHandler._
+
+ type Logger = (PrintWriter, Option[FSDataOutputStream])
+
+ private var logRoot: URI = _
+ private var fs: FileSystem = _
+ private val writers = HashMap.empty[String, Logger]
+
+ initialize()
+
+ override def apply(event: KyuubiEvent): Unit = {
+
+ val (writer, stream) = getOrUpdate(event)
+ // scalastyle:off println
+ writer.println(event.toJson)
+ // scalastyle:on println
+ writer.flush()
+ stream.foreach(_.hflush())
Review Comment:
Can we register a shutdownHook to ensure the file is closed ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] codecov-commenter commented on pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
codecov-commenter commented on PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#issuecomment-1208877188
# [Codecov](https://codecov.io/gh/apache/incubator-kyuubi/pull/3185?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#3185](https://codecov.io/gh/apache/incubator-kyuubi/pull/3185?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (18e48c9) into [master](https://codecov.io/gh/apache/incubator-kyuubi/commit/9d4d2948bb3e89c0166aff3927a7f3032093cea1?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (9d4d294) will **increase** coverage by `0.03%`.
> The diff coverage is `57.26%`.
```diff
@@ Coverage Diff @@
## master #3185 +/- ##
============================================
+ Coverage 51.52% 51.55% +0.03%
Complexity 6 6
============================================
Files 456 457 +1
Lines 25488 25718 +230
Branches 3543 3616 +73
============================================
+ Hits 13132 13260 +128
- Misses 11096 11155 +59
- Partials 1260 1303 +43
```
| [Impacted Files](https://codecov.io/gh/apache/incubator-kyuubi/pull/3185?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [...g/apache/kyuubi/engine/spark/KyuubiSparkUtil.scala](https://codecov.io/gh/apache/incubator-kyuubi/pull/3185/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-ZXh0ZXJuYWxzL2t5dXViaS1zcGFyay1zcWwtZW5naW5lL3NyYy9tYWluL3NjYWxhL29yZy9hcGFjaGUva3l1dWJpL2VuZ2luZS9zcGFyay9LeXV1YmlTcGFya1V0aWwuc2NhbGE=) | `65.51% <32.00%> (-25.40%)` | :arrow_down: |
| [...ache/spark/kyuubi/SparkSQLLineageParseHelper.scala](https://codecov.io/gh/apache/incubator-kyuubi/pull/3185/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-ZXh0ZXJuYWxzL2t5dXViaS1zcGFyay1zcWwtZW5naW5lL3NyYy9tYWluL3NjYWxhL29yZy9hcGFjaGUvc3Bhcmsva3l1dWJpL1NwYXJrU1FMTGluZWFnZVBhcnNlSGVscGVyLnNjYWxh) | `56.21% <56.21%> (ø)` | |
| [...uubi/engine/spark/operation/ExecuteStatement.scala](https://codecov.io/gh/apache/incubator-kyuubi/pull/3185/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-ZXh0ZXJuYWxzL2t5dXViaS1zcGFyay1zcWwtZW5naW5lL3NyYy9tYWluL3NjYWxhL29yZy9hcGFjaGUva3l1dWJpL2VuZ2luZS9zcGFyay9vcGVyYXRpb24vRXhlY3V0ZVN0YXRlbWVudC5zY2FsYQ==) | `84.84% <88.88%> (+0.23%)` | :arrow_up: |
| [...uubi/engine/spark/events/SparkOperationEvent.scala](https://codecov.io/gh/apache/incubator-kyuubi/pull/3185/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-ZXh0ZXJuYWxzL2t5dXViaS1zcGFyay1zcWwtZW5naW5lL3NyYy9tYWluL3NjYWxhL29yZy9hcGFjaGUva3l1dWJpL2VuZ2luZS9zcGFyay9ldmVudHMvU3BhcmtPcGVyYXRpb25FdmVudC5zY2FsYQ==) | `88.37% <90.00%> (-0.52%)` | :arrow_down: |
| [...in/scala/org/apache/kyuubi/config/KyuubiConf.scala](https://codecov.io/gh/apache/incubator-kyuubi/pull/3185/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-a3l1dWJpLWNvbW1vbi9zcmMvbWFpbi9zY2FsYS9vcmcvYXBhY2hlL2t5dXViaS9jb25maWcvS3l1dWJpQ29uZi5zY2FsYQ==) | `97.33% <100.00%> (+0.01%)` | :arrow_up: |
| [...apache/kyuubi/engine/JpsApplicationOperation.scala](https://codecov.io/gh/apache/incubator-kyuubi/pull/3185/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-a3l1dWJpLXNlcnZlci9zcmMvbWFpbi9zY2FsYS9vcmcvYXBhY2hlL2t5dXViaS9lbmdpbmUvSnBzQXBwbGljYXRpb25PcGVyYXRpb24uc2NhbGE=) | `77.41% <0.00%> (-3.23%)` | :arrow_down: |
| [...rg/apache/kyuubi/ctl/cmd/log/LogBatchCommand.scala](https://codecov.io/gh/apache/incubator-kyuubi/pull/3185/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-a3l1dWJpLWN0bC9zcmMvbWFpbi9zY2FsYS9vcmcvYXBhY2hlL2t5dXViaS9jdGwvY21kL2xvZy9Mb2dCYXRjaENvbW1hbmQuc2NhbGE=) | `78.00% <0.00%> (-2.00%)` | :arrow_down: |
| [...pache/kyuubi/engine/YarnApplicationOperation.scala](https://codecov.io/gh/apache/incubator-kyuubi/pull/3185/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-a3l1dWJpLXNlcnZlci9zcmMvbWFpbi9zY2FsYS9vcmcvYXBhY2hlL2t5dXViaS9lbmdpbmUvWWFybkFwcGxpY2F0aW9uT3BlcmF0aW9uLnNjYWxh) | `64.81% <0.00%> (-1.86%)` | :arrow_down: |
| [...yuubi/server/metadata/jdbc/JDBCMetadataStore.scala](https://codecov.io/gh/apache/incubator-kyuubi/pull/3185/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-a3l1dWJpLXNlcnZlci9zcmMvbWFpbi9zY2FsYS9vcmcvYXBhY2hlL2t5dXViaS9zZXJ2ZXIvbWV0YWRhdGEvamRiYy9KREJDTWV0YWRhdGFTdG9yZS5zY2FsYQ==) | `89.27% <0.00%> (-0.70%)` | :arrow_down: |
| [...n/scala/org/apache/kyuubi/engine/ProcBuilder.scala](https://codecov.io/gh/apache/incubator-kyuubi/pull/3185/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-a3l1dWJpLXNlcnZlci9zcmMvbWFpbi9zY2FsYS9vcmcvYXBhY2hlL2t5dXViaS9lbmdpbmUvUHJvY0J1aWxkZXIuc2NhbGE=) | `83.75% <0.00%> (+0.62%)` | :arrow_up: |
| ... and [2 more](https://codecov.io/gh/apache/incubator-kyuubi/pull/3185/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | |
:mega: Codecov can now indicate which changes are the most critical in Pull Requests. [Learn more](https://about.codecov.io/product/feature/runtime-insights/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] ulysses-you commented on a diff in pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
ulysses-you commented on code in PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#discussion_r941040683
##########
externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/spark/kyuubi/SparkSQLLineageParseHelper.scala:
##########
@@ -0,0 +1,366 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.kyuubi
+
+import scala.collection.immutable.ListMap
+import scala.util.{Failure, Success, Try}
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat, CatalogTable, HiveTableRelation}
+import org.apache.spark.sql.catalyst.expressions.{Alias, NamedExpression}
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.connector.catalog.{CatalogPlugin, Identifier, TableCatalog}
+import org.apache.spark.sql.execution.datasources.LogicalRelation
+import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, DataSourceV2ScanRelation}
+
+import org.apache.kyuubi.Logging
+import org.apache.kyuubi.engine.spark.KyuubiSparkUtil.isSparkVersionAtMost
+import org.apache.kyuubi.engine.spark.events.Lineage
+
+trait LineageParser {
+ def sparkSession: SparkSession
+
+ def parse(plan: LogicalPlan): Lineage = {
+ val ret = {
+ extractColumnsLineage(plan, ListMap()).toList.collect {
+ case (k, columns) =>
+ k.split('#').head -> columns.map(_.split('#').head)
+ }
+ }
+ val (inputTables, outputTables) = ret.foldLeft((List[String](), List[String]())) {
+ case ((inputs, outputs), (out, in)) =>
+ val x = (inputs ++ in.map(_.split('.').init.mkString("."))).filter(_.nonEmpty)
+ val y = outputs ++ List(out.split('.').init.mkString(".")).filter(_.nonEmpty)
+ (x, y)
+ }
+ Lineage(inputTables.distinct, outputTables.distinct, ret)
+ }
+
+ private def mergeColumnsLineage(
+ left: ListMap[String, List[String]],
+ right: ListMap[String, List[String]]): ListMap[String, List[String]] = {
+ if (left.isEmpty || right.isEmpty) left ++ right
+ else {
+ val head =
+ ListMap(left.head._1 -> (left.head._2 ++ right.getOrElse(left.head._1, List.empty)))
+ head ++ mergeColumnsLineage(left.tail, right - left.head._1)
+ }
+ }
+
+ private def joinColumnsLineage(
+ parent: ListMap[String, List[String]],
+ child: ListMap[String, List[String]]): ListMap[String, List[String]] = {
+ if (parent.isEmpty) child
+ else {
+ val childTmp = child.map { case (k, v) => (k.split('#').last, v) }
+ parent.map { case (k, columns) =>
+ k -> {
+ columns.flatMap {
+ case col if childTmp.contains(col.split('#').last) => childTmp(col.split('#').last)
+ case _ => List()
+ }
+ }
+ }
+ }
+ }
+
+ private def extractColumnsLineage(
+ plan: LogicalPlan,
+ parentColumnsLineage: ListMap[String, List[String]]): ListMap[String, List[String]] = {
+
+ def getPlanField[T](field: String): T = {
+ getFieldVal[T](plan, field)
+ }
+
+ def getCurrentPlanField[T](curPlan: LogicalPlan, field: String): T = {
+ getFieldVal[T](curPlan, field)
+ }
+
+ def getPlanMethod[T](name: String): T = {
+ getMethod[T](plan, name)
+ }
+
+ def getQuery: LogicalPlan = {
+ getPlanField[LogicalPlan]("query")
+ }
+
+ plan match {
+ // For command
+ case p if p.nodeName == "AlterViewAsCommand" =>
+ val analyzed = sparkSession.sessionState.analyzer.execute(getQuery)
+ val view = getPlanField[TableIdentifier]("name").unquotedString
+ extractColumnsLineage(analyzed, parentColumnsLineage).map { case (k, v) =>
+ (s"$view.$k", v)
+ }
+
+ case p if p.nodeName == "CreateViewCommand" =>
+ val view = getPlanField[TableIdentifier]("name").unquotedString
+ val outputCols =
+ getPlanField[Seq[(String, Option[String])]]("userSpecifiedColumns").map(_._1)
+ val query =
+ if (isSparkVersionAtMost("3.1")) {
+ getPlanField[LogicalPlan]("child")
+ } else {
+ getPlanField[LogicalPlan]("plan")
+ }
+ val analyzed = sparkSession.sessionState.analyzer.execute(query)
+ extractColumnsLineage(analyzed, parentColumnsLineage).zipWithIndex.map {
+ case (x, i) if outputCols.nonEmpty => s"$view.${outputCols(i)}" -> x._2
+ case ((k, v), _) => s"$view.$k" -> v
+ }
+
+ case p if p.nodeName == "CreateDataSourceTableAsSelectCommand" =>
+ val table = getPlanField[CatalogTable]("table").qualifiedName
+ val analyzed = sparkSession.sessionState.analyzer.execute(getQuery)
+ extractColumnsLineage(analyzed, parentColumnsLineage).map { case (k, v) =>
+ s"$table.$k" -> v
+ }
+
+ case p
+ if p.nodeName == "CreateHiveTableAsSelectCommand" ||
+ p.nodeName == "OptimizedCreateHiveTableAsSelectCommand" =>
+ val table = getPlanField[CatalogTable]("tableDesc").qualifiedName
+ extractColumnsLineage(getQuery, parentColumnsLineage).map { case (k, v) =>
+ s"$table.$k" -> v
+ }
+
+ case p
+ if p.nodeName == "CreateTableAsSelect" ||
+ p.nodeName == "ReplaceTableAsSelect" =>
+ val (table, namespace, catalog) =
+ if (isSparkVersionAtMost("3.2")) {
+ (
+ getPlanField[Identifier]("tableName").name,
+ getPlanField[Identifier]("tableName").namespace.mkString("."),
+ getPlanField[TableCatalog]("catalog").name())
+ } else {
+ (
+ getPlanMethod[Identifier]("tableName").name(),
+ getPlanMethod[Identifier]("tableName").namespace().mkString("."),
+ getCurrentPlanField[CatalogPlugin](
+ getPlanMethod[LogicalPlan]("left"),
+ "catalog").name())
+ }
+ extractColumnsLineage(getQuery, parentColumnsLineage).map { case (k, v) =>
+ s"$catalog.$namespace.$table.$k" -> v
+ }
+
+ case p if p.nodeName == "InsertIntoDataSourceCommand" =>
+ val logicalRelation = getPlanField[LogicalRelation]("logicalRelation")
+ val table = logicalRelation.catalogTable.map(_.qualifiedName).getOrElse("")
+ extractColumnsLineage(getQuery, parentColumnsLineage).map { case (k, v) =>
+ s"$table.$k" -> v
+ }
+
+ case p if p.nodeName == "InsertIntoHadoopFsRelationCommand" =>
+ val table =
+ getPlanField[Option[CatalogTable]]("catalogTable").map(_.qualifiedName).getOrElse("")
+ extractColumnsLineage(getQuery, parentColumnsLineage).map { case (k, v) =>
+ s"$table.$k" -> v
+ }
+
+ case p
+ if p.nodeName == "InsertIntoDataSourceDirCommand" ||
+ p.nodeName == "InsertIntoHiveDirCommand" =>
+ val dir =
+ getPlanField[CatalogStorageFormat]("storage").locationUri.map(_.toString).getOrElse("")
+ extractColumnsLineage(getQuery, parentColumnsLineage).map { case (k, v) =>
+ s"`$dir`.$k" -> v
+ }
+
+ case p if p.nodeName == "InsertIntoHiveTable" =>
+ val table = getPlanField[CatalogTable]("table").qualifiedName
+ extractColumnsLineage(getQuery, parentColumnsLineage).map { case (k, v) =>
+ s"$table.$k" -> v
+ }
+
+ case p if p.nodeName == "SaveIntoDataSourceCommand" =>
+ // TODO: Should get the table via datasource options?
+ extractColumnsLineage(getQuery, parentColumnsLineage)
+
+ // For query
+ case p: Project =>
+ val nextColumnsLineage =
+ joinColumnsLineage(parentColumnsLineage, getSelectColumnLineage(p.projectList))
+ p.children.map(extractColumnsLineage(_, nextColumnsLineage)).reduce(mergeColumnsLineage)
+
+ case p: Aggregate =>
+ val nextColumnsLineage =
+ joinColumnsLineage(parentColumnsLineage, getSelectColumnLineage(p.aggregateExpressions))
+ p.children.map(extractColumnsLineage(_, nextColumnsLineage)).reduce(mergeColumnsLineage)
+
+ case p: Union =>
+ // merge all children in to one derivedColumns
+ val childrenUnion =
+ p.children.map(extractColumnsLineage(_, ListMap())).map(_.values).reduce {
+ (left, right) =>
+ left.zip(right).map(col => (col._1 ++ col._2).distinct)
+ }
+ val childrenColumnsLineage =
+ ListMap(p.output.map(a => s"${a.name}#${a.exprId.id}").toList.zip(childrenUnion): _*)
+ joinColumnsLineage(parentColumnsLineage, childrenColumnsLineage)
+
+ case p: SubqueryAlias =>
+ extractColumnsLineage(p.child, parentColumnsLineage)
Review Comment:
it seems never reach here with optimized plan
##########
externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/spark/kyuubi/SparkSQLLineageParseHelper.scala:
##########
@@ -0,0 +1,366 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.kyuubi
+
+import scala.collection.immutable.ListMap
+import scala.util.{Failure, Success, Try}
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat, CatalogTable, HiveTableRelation}
+import org.apache.spark.sql.catalyst.expressions.{Alias, NamedExpression}
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.connector.catalog.{CatalogPlugin, Identifier, TableCatalog}
+import org.apache.spark.sql.execution.datasources.LogicalRelation
+import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, DataSourceV2ScanRelation}
+
+import org.apache.kyuubi.Logging
+import org.apache.kyuubi.engine.spark.KyuubiSparkUtil.isSparkVersionAtMost
+import org.apache.kyuubi.engine.spark.events.Lineage
+
+trait LineageParser {
+ def sparkSession: SparkSession
+
+ def parse(plan: LogicalPlan): Lineage = {
+ val ret = {
+ extractColumnsLineage(plan, ListMap()).toList.collect {
+ case (k, columns) =>
+ k.split('#').head -> columns.map(_.split('#').head)
+ }
+ }
+ val (inputTables, outputTables) = ret.foldLeft((List[String](), List[String]())) {
+ case ((inputs, outputs), (out, in)) =>
+ val x = (inputs ++ in.map(_.split('.').init.mkString("."))).filter(_.nonEmpty)
+ val y = outputs ++ List(out.split('.').init.mkString(".")).filter(_.nonEmpty)
+ (x, y)
+ }
+ Lineage(inputTables.distinct, outputTables.distinct, ret)
+ }
+
+ private def mergeColumnsLineage(
+ left: ListMap[String, List[String]],
+ right: ListMap[String, List[String]]): ListMap[String, List[String]] = {
+ if (left.isEmpty || right.isEmpty) left ++ right
+ else {
+ val head =
+ ListMap(left.head._1 -> (left.head._2 ++ right.getOrElse(left.head._1, List.empty)))
+ head ++ mergeColumnsLineage(left.tail, right - left.head._1)
+ }
+ }
+
+ private def joinColumnsLineage(
+ parent: ListMap[String, List[String]],
+ child: ListMap[String, List[String]]): ListMap[String, List[String]] = {
+ if (parent.isEmpty) child
+ else {
+ val childTmp = child.map { case (k, v) => (k.split('#').last, v) }
+ parent.map { case (k, columns) =>
+ k -> {
+ columns.flatMap {
+ case col if childTmp.contains(col.split('#').last) => childTmp(col.split('#').last)
+ case _ => List()
+ }
+ }
+ }
+ }
+ }
+
+ private def extractColumnsLineage(
+ plan: LogicalPlan,
+ parentColumnsLineage: ListMap[String, List[String]]): ListMap[String, List[String]] = {
+
+ def getPlanField[T](field: String): T = {
+ getFieldVal[T](plan, field)
+ }
+
+ def getCurrentPlanField[T](curPlan: LogicalPlan, field: String): T = {
+ getFieldVal[T](curPlan, field)
+ }
+
+ def getPlanMethod[T](name: String): T = {
+ getMethod[T](plan, name)
+ }
+
+ def getQuery: LogicalPlan = {
+ getPlanField[LogicalPlan]("query")
+ }
+
+ plan match {
+ // For command
+ case p if p.nodeName == "AlterViewAsCommand" =>
+ val analyzed = sparkSession.sessionState.analyzer.execute(getQuery)
+ val view = getPlanField[TableIdentifier]("name").unquotedString
+ extractColumnsLineage(analyzed, parentColumnsLineage).map { case (k, v) =>
+ (s"$view.$k", v)
+ }
+
+ case p if p.nodeName == "CreateViewCommand" =>
+ val view = getPlanField[TableIdentifier]("name").unquotedString
+ val outputCols =
+ getPlanField[Seq[(String, Option[String])]]("userSpecifiedColumns").map(_._1)
+ val query =
+ if (isSparkVersionAtMost("3.1")) {
+ getPlanField[LogicalPlan]("child")
+ } else {
+ getPlanField[LogicalPlan]("plan")
+ }
+ val analyzed = sparkSession.sessionState.analyzer.execute(query)
+ extractColumnsLineage(analyzed, parentColumnsLineage).zipWithIndex.map {
+ case (x, i) if outputCols.nonEmpty => s"$view.${outputCols(i)}" -> x._2
+ case ((k, v), _) => s"$view.$k" -> v
+ }
+
+ case p if p.nodeName == "CreateDataSourceTableAsSelectCommand" =>
+ val table = getPlanField[CatalogTable]("table").qualifiedName
+ val analyzed = sparkSession.sessionState.analyzer.execute(getQuery)
+ extractColumnsLineage(analyzed, parentColumnsLineage).map { case (k, v) =>
+ s"$table.$k" -> v
+ }
+
+ case p
+ if p.nodeName == "CreateHiveTableAsSelectCommand" ||
+ p.nodeName == "OptimizedCreateHiveTableAsSelectCommand" =>
+ val table = getPlanField[CatalogTable]("tableDesc").qualifiedName
+ extractColumnsLineage(getQuery, parentColumnsLineage).map { case (k, v) =>
+ s"$table.$k" -> v
+ }
+
+ case p
+ if p.nodeName == "CreateTableAsSelect" ||
+ p.nodeName == "ReplaceTableAsSelect" =>
+ val (table, namespace, catalog) =
+ if (isSparkVersionAtMost("3.2")) {
+ (
+ getPlanField[Identifier]("tableName").name,
+ getPlanField[Identifier]("tableName").namespace.mkString("."),
+ getPlanField[TableCatalog]("catalog").name())
+ } else {
+ (
+ getPlanMethod[Identifier]("tableName").name(),
+ getPlanMethod[Identifier]("tableName").namespace().mkString("."),
+ getCurrentPlanField[CatalogPlugin](
+ getPlanMethod[LogicalPlan]("left"),
+ "catalog").name())
+ }
+ extractColumnsLineage(getQuery, parentColumnsLineage).map { case (k, v) =>
+ s"$catalog.$namespace.$table.$k" -> v
+ }
+
+ case p if p.nodeName == "InsertIntoDataSourceCommand" =>
+ val logicalRelation = getPlanField[LogicalRelation]("logicalRelation")
+ val table = logicalRelation.catalogTable.map(_.qualifiedName).getOrElse("")
+ extractColumnsLineage(getQuery, parentColumnsLineage).map { case (k, v) =>
+ s"$table.$k" -> v
+ }
+
+ case p if p.nodeName == "InsertIntoHadoopFsRelationCommand" =>
+ val table =
+ getPlanField[Option[CatalogTable]]("catalogTable").map(_.qualifiedName).getOrElse("")
+ extractColumnsLineage(getQuery, parentColumnsLineage).map { case (k, v) =>
+ s"$table.$k" -> v
+ }
+
+ case p
+ if p.nodeName == "InsertIntoDataSourceDirCommand" ||
+ p.nodeName == "InsertIntoHiveDirCommand" =>
+ val dir =
+ getPlanField[CatalogStorageFormat]("storage").locationUri.map(_.toString).getOrElse("")
+ extractColumnsLineage(getQuery, parentColumnsLineage).map { case (k, v) =>
+ s"`$dir`.$k" -> v
+ }
+
+ case p if p.nodeName == "InsertIntoHiveTable" =>
+ val table = getPlanField[CatalogTable]("table").qualifiedName
+ extractColumnsLineage(getQuery, parentColumnsLineage).map { case (k, v) =>
+ s"$table.$k" -> v
+ }
+
+ case p if p.nodeName == "SaveIntoDataSourceCommand" =>
+ // TODO: Should get the table via datasource options?
+ extractColumnsLineage(getQuery, parentColumnsLineage)
+
+ // For query
+ case p: Project =>
+ val nextColumnsLineage =
+ joinColumnsLineage(parentColumnsLineage, getSelectColumnLineage(p.projectList))
+ p.children.map(extractColumnsLineage(_, nextColumnsLineage)).reduce(mergeColumnsLineage)
+
+ case p: Aggregate =>
+ val nextColumnsLineage =
+ joinColumnsLineage(parentColumnsLineage, getSelectColumnLineage(p.aggregateExpressions))
+ p.children.map(extractColumnsLineage(_, nextColumnsLineage)).reduce(mergeColumnsLineage)
+
+ case p: Union =>
+ // merge all children in to one derivedColumns
+ val childrenUnion =
+ p.children.map(extractColumnsLineage(_, ListMap())).map(_.values).reduce {
+ (left, right) =>
+ left.zip(right).map(col => (col._1 ++ col._2).distinct)
+ }
+ val childrenColumnsLineage =
+ ListMap(p.output.map(a => s"${a.name}#${a.exprId.id}").toList.zip(childrenUnion): _*)
+ joinColumnsLineage(parentColumnsLineage, childrenColumnsLineage)
+
+ case p: SubqueryAlias =>
+ extractColumnsLineage(p.child, parentColumnsLineage)
+
+ case p: LogicalRelation if p.catalogTable.nonEmpty =>
+ val tableName = p.catalogTable.get.qualifiedName
+ val dataColIds = p.output.map(_.exprId.id.toString).toSet
+ val dataColNames = p.output.map(_.name)
+ if (parentColumnsLineage.nonEmpty) {
+ parentColumnsLineage.map {
+ case (k, columns) =>
+ k -> columns.collect {
+ case col if dataColIds.contains(col.split('#').last) => s"$tableName.$col"
+ }
+ }
+ } else {
+ val colPair = dataColNames.map { col =>
+ col -> List(s"$tableName.$col")
+ }
+ ListMap(colPair: _*)
+ }
+
+ case p: HiveTableRelation =>
+ val tableName = p.tableMeta.qualifiedName
+ val dataCols = p.dataCols.map(_.exprId.id.toString).toSet
+ val dataColNames = p.output.map(_.name)
+ if (parentColumnsLineage.nonEmpty) {
+ parentColumnsLineage.map { case (k, columns) =>
+ k -> columns.collect {
+ case col if dataCols.contains(col.split('#').last) => s"$tableName.$col"
+ }
+ }
+ } else {
+ val colPair = dataColNames.map { col =>
+ col -> List(s"$tableName.$col")
+ }
+ ListMap(colPair: _*)
+ }
+
+ case p: DataSourceV2Relation =>
+ val tableName = p.name
+ val dataCols = p.output.map(_.exprId.id.toString).toSet
+ val dataColNames = p.output.map(_.name)
+ if (parentColumnsLineage.nonEmpty) {
+ parentColumnsLineage.map { case (k, columns) =>
+ k -> columns.collect {
+ case col if dataCols.contains(col.split('#').last) => s"$tableName.$col"
+ }
+ }
+ } else {
+ val colPair = dataColNames.map { col =>
+ col -> List(s"$tableName.$col")
+ }
+ ListMap(colPair: _*)
+ }
+
+ case p: DataSourceV2ScanRelation =>
+ val tableName = p.name
+ val dataCols = p.output.map(_.exprId.id.toString).toSet
+ val dataColNames = p.output.map(_.name)
+ if (parentColumnsLineage.nonEmpty) {
+ parentColumnsLineage.map { case (k, columns) =>
+ k -> columns.collect {
+ case col if dataCols.contains(col.split('#').last) => s"$tableName.$col"
+ }
+ }
+ } else {
+ val colPair = dataColNames.map { col =>
+ col -> List(s"$tableName.$col")
+ }
+ ListMap(colPair: _*)
+ }
+
+ case p: LocalRelation =>
+ val dataCols = p.output.map(_.exprId.id.toString).toSet
+ val dataColNames = p.output.map(_.name)
+ if (parentColumnsLineage.nonEmpty) {
+ parentColumnsLineage.map { case (k, columns) =>
+ k -> columns.collect {
+ case col if dataCols.contains(col.split('#').last) => col
+ }
+ }
+ } else {
+ val colPair = dataColNames.map { col =>
+ col -> List(col)
+ }
+ ListMap(colPair: _*)
+ }
+
+ case p if p.children.isEmpty => ListMap.empty
+ case p =>
+ p.children.map(extractColumnsLineage(_, parentColumnsLineage)).reduce(mergeColumnsLineage)
+ }
+ }
+
+ private def getSelectColumnLineage(exps: Seq[NamedExpression]): ListMap[String, List[String]] = {
+ val result = exps.map { exp =>
+ val derivedColumns = exp.references.map(e => s"${e.name}#${e.exprId.id}").toList
+ exp match {
+ case e: Alias =>
+ (s"${e.name}#${e.exprId.id}" -> derivedColumns)
+ case _ =>
+ (exp.toString -> derivedColumns)
+ }
Review Comment:
can we use `AttributeMap` instead of `ListMap[String, List[String]]` ? for example:
```scala
def mergeReference(
left: AttributeMap[AttributeSet],
right: AttributeMap[AttributeSet]): AttributeMap[AttributeSet] = {
AttributeMap(
left ++ right.map { case (k, attrs) =>
k -> (attrs ++ right.getOrElse(k, AttributeSet.empty))
}
)
}
def replaceParentReferenceByChildAlias(
parent: AttributeMap[AttributeSet],
child: AttributeMap[AttributeSet]): AttributeMap[AttributeSet] = {
AttributeMap(
parent.map { case (k, attrs) =>
k -> AttributeSet(attrs.flatMap(attr => child.get(attr).getOrElse(AttributeSet(attr))))
}
)
}
private def selectAttributeReferenceMap(
named: Seq[NamedExpression]): AttributeMap[AttributeSet] = {
AttributeMap.apply(
named.map {
case a: Alias => a.toAttribute -> a.references
case a: Attribute => a -> a.references
}
)
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] iodone commented on a diff in pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
iodone commented on code in PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#discussion_r941087333
##########
externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/spark/kyuubi/SparkSQLLineageParseHelper.scala:
##########
@@ -0,0 +1,366 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.kyuubi
+
+import scala.collection.immutable.ListMap
+import scala.util.{Failure, Success, Try}
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat, CatalogTable, HiveTableRelation}
+import org.apache.spark.sql.catalyst.expressions.{Alias, NamedExpression}
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.connector.catalog.{CatalogPlugin, Identifier, TableCatalog}
+import org.apache.spark.sql.execution.datasources.LogicalRelation
+import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, DataSourceV2ScanRelation}
+
+import org.apache.kyuubi.Logging
+import org.apache.kyuubi.engine.spark.KyuubiSparkUtil.isSparkVersionAtMost
+import org.apache.kyuubi.engine.spark.events.Lineage
+
+trait LineageParser {
+ def sparkSession: SparkSession
+
+ def parse(plan: LogicalPlan): Lineage = {
+ val ret = {
+ extractColumnsLineage(plan, ListMap()).toList.collect {
+ case (k, columns) =>
+ k.split('#').head -> columns.map(_.split('#').head)
+ }
+ }
+ val (inputTables, outputTables) = ret.foldLeft((List[String](), List[String]())) {
+ case ((inputs, outputs), (out, in)) =>
+ val x = (inputs ++ in.map(_.split('.').init.mkString("."))).filter(_.nonEmpty)
+ val y = outputs ++ List(out.split('.').init.mkString(".")).filter(_.nonEmpty)
+ (x, y)
+ }
+ Lineage(inputTables.distinct, outputTables.distinct, ret)
+ }
+
+ private def mergeColumnsLineage(
+ left: ListMap[String, List[String]],
+ right: ListMap[String, List[String]]): ListMap[String, List[String]] = {
+ if (left.isEmpty || right.isEmpty) left ++ right
+ else {
+ val head =
+ ListMap(left.head._1 -> (left.head._2 ++ right.getOrElse(left.head._1, List.empty)))
+ head ++ mergeColumnsLineage(left.tail, right - left.head._1)
+ }
+ }
+
+ private def joinColumnsLineage(
+ parent: ListMap[String, List[String]],
+ child: ListMap[String, List[String]]): ListMap[String, List[String]] = {
+ if (parent.isEmpty) child
+ else {
+ val childTmp = child.map { case (k, v) => (k.split('#').last, v) }
+ parent.map { case (k, columns) =>
+ k -> {
+ columns.flatMap {
+ case col if childTmp.contains(col.split('#').last) => childTmp(col.split('#').last)
+ case _ => List()
+ }
+ }
+ }
+ }
+ }
+
+ private def extractColumnsLineage(
+ plan: LogicalPlan,
+ parentColumnsLineage: ListMap[String, List[String]]): ListMap[String, List[String]] = {
+
+ def getPlanField[T](field: String): T = {
+ getFieldVal[T](plan, field)
+ }
+
+ def getCurrentPlanField[T](curPlan: LogicalPlan, field: String): T = {
+ getFieldVal[T](curPlan, field)
+ }
+
+ def getPlanMethod[T](name: String): T = {
+ getMethod[T](plan, name)
+ }
+
+ def getQuery: LogicalPlan = {
+ getPlanField[LogicalPlan]("query")
+ }
+
+ plan match {
+ // For command
+ case p if p.nodeName == "AlterViewAsCommand" =>
+ val analyzed = sparkSession.sessionState.analyzer.execute(getQuery)
+ val view = getPlanField[TableIdentifier]("name").unquotedString
+ extractColumnsLineage(analyzed, parentColumnsLineage).map { case (k, v) =>
+ (s"$view.$k", v)
+ }
+
+ case p if p.nodeName == "CreateViewCommand" =>
+ val view = getPlanField[TableIdentifier]("name").unquotedString
+ val outputCols =
+ getPlanField[Seq[(String, Option[String])]]("userSpecifiedColumns").map(_._1)
+ val query =
+ if (isSparkVersionAtMost("3.1")) {
+ getPlanField[LogicalPlan]("child")
+ } else {
+ getPlanField[LogicalPlan]("plan")
+ }
+ val analyzed = sparkSession.sessionState.analyzer.execute(query)
+ extractColumnsLineage(analyzed, parentColumnsLineage).zipWithIndex.map {
+ case (x, i) if outputCols.nonEmpty => s"$view.${outputCols(i)}" -> x._2
+ case ((k, v), _) => s"$view.$k" -> v
+ }
+
+ case p if p.nodeName == "CreateDataSourceTableAsSelectCommand" =>
+ val table = getPlanField[CatalogTable]("table").qualifiedName
+ val analyzed = sparkSession.sessionState.analyzer.execute(getQuery)
+ extractColumnsLineage(analyzed, parentColumnsLineage).map { case (k, v) =>
+ s"$table.$k" -> v
+ }
+
+ case p
+ if p.nodeName == "CreateHiveTableAsSelectCommand" ||
+ p.nodeName == "OptimizedCreateHiveTableAsSelectCommand" =>
+ val table = getPlanField[CatalogTable]("tableDesc").qualifiedName
+ extractColumnsLineage(getQuery, parentColumnsLineage).map { case (k, v) =>
+ s"$table.$k" -> v
+ }
+
+ case p
+ if p.nodeName == "CreateTableAsSelect" ||
+ p.nodeName == "ReplaceTableAsSelect" =>
+ val (table, namespace, catalog) =
+ if (isSparkVersionAtMost("3.2")) {
+ (
+ getPlanField[Identifier]("tableName").name,
+ getPlanField[Identifier]("tableName").namespace.mkString("."),
+ getPlanField[TableCatalog]("catalog").name())
+ } else {
+ (
+ getPlanMethod[Identifier]("tableName").name(),
+ getPlanMethod[Identifier]("tableName").namespace().mkString("."),
+ getCurrentPlanField[CatalogPlugin](
+ getPlanMethod[LogicalPlan]("left"),
+ "catalog").name())
+ }
+ extractColumnsLineage(getQuery, parentColumnsLineage).map { case (k, v) =>
+ s"$catalog.$namespace.$table.$k" -> v
+ }
+
+ case p if p.nodeName == "InsertIntoDataSourceCommand" =>
+ val logicalRelation = getPlanField[LogicalRelation]("logicalRelation")
+ val table = logicalRelation.catalogTable.map(_.qualifiedName).getOrElse("")
+ extractColumnsLineage(getQuery, parentColumnsLineage).map { case (k, v) =>
+ s"$table.$k" -> v
+ }
+
+ case p if p.nodeName == "InsertIntoHadoopFsRelationCommand" =>
+ val table =
+ getPlanField[Option[CatalogTable]]("catalogTable").map(_.qualifiedName).getOrElse("")
+ extractColumnsLineage(getQuery, parentColumnsLineage).map { case (k, v) =>
+ s"$table.$k" -> v
+ }
+
+ case p
+ if p.nodeName == "InsertIntoDataSourceDirCommand" ||
+ p.nodeName == "InsertIntoHiveDirCommand" =>
+ val dir =
+ getPlanField[CatalogStorageFormat]("storage").locationUri.map(_.toString).getOrElse("")
+ extractColumnsLineage(getQuery, parentColumnsLineage).map { case (k, v) =>
+ s"`$dir`.$k" -> v
+ }
+
+ case p if p.nodeName == "InsertIntoHiveTable" =>
+ val table = getPlanField[CatalogTable]("table").qualifiedName
+ extractColumnsLineage(getQuery, parentColumnsLineage).map { case (k, v) =>
+ s"$table.$k" -> v
+ }
+
+ case p if p.nodeName == "SaveIntoDataSourceCommand" =>
+ // TODO: Should get the table via datasource options?
+ extractColumnsLineage(getQuery, parentColumnsLineage)
+
+ // For query
+ case p: Project =>
+ val nextColumnsLineage =
+ joinColumnsLineage(parentColumnsLineage, getSelectColumnLineage(p.projectList))
+ p.children.map(extractColumnsLineage(_, nextColumnsLineage)).reduce(mergeColumnsLineage)
+
+ case p: Aggregate =>
+ val nextColumnsLineage =
+ joinColumnsLineage(parentColumnsLineage, getSelectColumnLineage(p.aggregateExpressions))
+ p.children.map(extractColumnsLineage(_, nextColumnsLineage)).reduce(mergeColumnsLineage)
+
+ case p: Union =>
+ // merge all children in to one derivedColumns
+ val childrenUnion =
+ p.children.map(extractColumnsLineage(_, ListMap())).map(_.values).reduce {
+ (left, right) =>
+ left.zip(right).map(col => (col._1 ++ col._2).distinct)
+ }
+ val childrenColumnsLineage =
+ ListMap(p.output.map(a => s"${a.name}#${a.exprId.id}").toList.zip(childrenUnion): _*)
+ joinColumnsLineage(parentColumnsLineage, childrenColumnsLineage)
+
+ case p: SubqueryAlias =>
+ extractColumnsLineage(p.child, parentColumnsLineage)
+
+ case p: LogicalRelation if p.catalogTable.nonEmpty =>
+ val tableName = p.catalogTable.get.qualifiedName
+ val dataColIds = p.output.map(_.exprId.id.toString).toSet
+ val dataColNames = p.output.map(_.name)
+ if (parentColumnsLineage.nonEmpty) {
+ parentColumnsLineage.map {
+ case (k, columns) =>
+ k -> columns.collect {
+ case col if dataColIds.contains(col.split('#').last) => s"$tableName.$col"
+ }
+ }
+ } else {
+ val colPair = dataColNames.map { col =>
+ col -> List(s"$tableName.$col")
+ }
+ ListMap(colPair: _*)
+ }
+
+ case p: HiveTableRelation =>
+ val tableName = p.tableMeta.qualifiedName
+ val dataCols = p.dataCols.map(_.exprId.id.toString).toSet
+ val dataColNames = p.output.map(_.name)
+ if (parentColumnsLineage.nonEmpty) {
+ parentColumnsLineage.map { case (k, columns) =>
+ k -> columns.collect {
+ case col if dataCols.contains(col.split('#').last) => s"$tableName.$col"
+ }
+ }
+ } else {
+ val colPair = dataColNames.map { col =>
+ col -> List(s"$tableName.$col")
+ }
+ ListMap(colPair: _*)
+ }
+
+ case p: DataSourceV2Relation =>
+ val tableName = p.name
+ val dataCols = p.output.map(_.exprId.id.toString).toSet
+ val dataColNames = p.output.map(_.name)
+ if (parentColumnsLineage.nonEmpty) {
+ parentColumnsLineage.map { case (k, columns) =>
+ k -> columns.collect {
+ case col if dataCols.contains(col.split('#').last) => s"$tableName.$col"
+ }
+ }
+ } else {
+ val colPair = dataColNames.map { col =>
+ col -> List(s"$tableName.$col")
+ }
+ ListMap(colPair: _*)
+ }
+
+ case p: DataSourceV2ScanRelation =>
+ val tableName = p.name
+ val dataCols = p.output.map(_.exprId.id.toString).toSet
+ val dataColNames = p.output.map(_.name)
+ if (parentColumnsLineage.nonEmpty) {
+ parentColumnsLineage.map { case (k, columns) =>
+ k -> columns.collect {
+ case col if dataCols.contains(col.split('#').last) => s"$tableName.$col"
+ }
+ }
+ } else {
+ val colPair = dataColNames.map { col =>
+ col -> List(s"$tableName.$col")
+ }
+ ListMap(colPair: _*)
+ }
+
+ case p: LocalRelation =>
+ val dataCols = p.output.map(_.exprId.id.toString).toSet
+ val dataColNames = p.output.map(_.name)
+ if (parentColumnsLineage.nonEmpty) {
+ parentColumnsLineage.map { case (k, columns) =>
+ k -> columns.collect {
+ case col if dataCols.contains(col.split('#').last) => col
+ }
+ }
+ } else {
+ val colPair = dataColNames.map { col =>
+ col -> List(col)
+ }
+ ListMap(colPair: _*)
+ }
+
+ case p if p.children.isEmpty => ListMap.empty
+ case p =>
+ p.children.map(extractColumnsLineage(_, parentColumnsLineage)).reduce(mergeColumnsLineage)
+ }
+ }
+
+ private def getSelectColumnLineage(exps: Seq[NamedExpression]): ListMap[String, List[String]] = {
+ val result = exps.map { exp =>
+ val derivedColumns = exp.references.map(e => s"${e.name}#${e.exprId.id}").toList
+ exp match {
+ case e: Alias =>
+ (s"${e.name}#${e.exprId.id}" -> derivedColumns)
+ case _ =>
+ (exp.toString -> derivedColumns)
+ }
Review Comment:
Thanks for the very good advice, I am ready to change it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] pan3793 commented on a diff in pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
pan3793 commented on code in PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#discussion_r952427086
##########
extensions/spark/kyuubi-spark-lineage/src/main/scala/org/apache/kyuubi/plugin/lineage/events/OperationLineageEvent.scala:
##########
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.lineage.events
+
+import org.apache.kyuubi.Utils
+import org.apache.kyuubi.events.KyuubiEvent
+
+case class ColumnLineage(column: String, ordinal: Int, originalColumns: Set[String])
+
+/**
+ * @param inputTables the tables of the operation will read
+ * @param outputTables the tables of the operation will write
+ * @param columnLineage the output columns are associated to columns of the real table's columns
+ */
+
+class Lineage(
+ val inputTables: List[String],
+ val outputTables: List[String],
+ val columnLineage: List[ColumnLineage]) {
+
+ override def equals(other: Any): Boolean = other match {
+ case otherLineage: Lineage =>
+ otherLineage.inputTables == inputTables && otherLineage.outputTables == outputTables &&
+ otherLineage.columnLineage == columnLineage
+ case _ => false
+ }
+
+ override def hashCode(): Int = super.hashCode()
+}
+
+object Lineage {
+ def apply(
+ inputTables: List[String],
+ outputTables: List[String],
+ columnLineage: List[(String, Set[String])]): Lineage = {
Review Comment:
looks redundant, how about construct a case class directly?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] pan3793 commented on a diff in pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
pan3793 commented on code in PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#discussion_r952426145
##########
extensions/spark/kyuubi-spark-lineage/src/main/scala/org/apache/kyuubi/plugin/lineage/events/OperationLineageEvent.scala:
##########
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.lineage.events
+
+import org.apache.kyuubi.Utils
+import org.apache.kyuubi.events.KyuubiEvent
+
+case class ColumnLineage(column: String, ordinal: Int, originalColumns: Set[String])
+
+/**
+ * @param inputTables the tables of the operation will read
+ * @param outputTables the tables of the operation will write
+ * @param columnLineage the output columns are associated to columns of the real table's columns
+ */
+
+class Lineage(
+ val inputTables: List[String],
+ val outputTables: List[String],
+ val columnLineage: List[ColumnLineage]) {
+
+ override def equals(other: Any): Boolean = other match {
Review Comment:
why not use case class?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] iodone commented on a diff in pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
iodone commented on code in PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#discussion_r952488098
##########
extensions/spark/kyuubi-spark-lineage/src/main/scala/org/apache/kyuubi/plugin/lineage/events/OperationLineageEvent.scala:
##########
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.lineage.events
+
+import org.apache.kyuubi.Utils
+import org.apache.kyuubi.events.KyuubiEvent
+
+case class ColumnLineage(column: String, ordinal: Int, originalColumns: Set[String])
+
+/**
+ * @param inputTables the tables of the operation will read
+ * @param outputTables the tables of the operation will write
+ * @param columnLineage the output columns are associated to columns of the real table's columns
+ */
+
+class Lineage(
+ val inputTables: List[String],
+ val outputTables: List[String],
+ val columnLineage: List[ColumnLineage]) {
+
+ override def equals(other: Any): Boolean = other match {
+ case otherLineage: Lineage =>
+ otherLineage.inputTables == inputTables && otherLineage.outputTables == outputTables &&
+ otherLineage.columnLineage == columnLineage
+ case _ => false
+ }
+
+ override def hashCode(): Int = super.hashCode()
+}
+
+object Lineage {
+ def apply(
+ inputTables: List[String],
+ outputTables: List[String],
+ columnLineage: List[(String, Set[String])]): Lineage = {
Review Comment:
Not redunant, this not use case class, this for compatibility with test writing. Otherwise, I need to change a lot of test code.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] cfmcgrady commented on a diff in pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
cfmcgrady commented on code in PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#discussion_r952335381
##########
extensions/spark/kyuubi-spark-lineage/src/test/scala/org/apache/kyuubi/plugin/lineage/helper/SparkSQLLineageParserHelperSuite.scala:
##########
@@ -0,0 +1,722 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.plugin.lineage.helper
+
+import scala.collection.immutable.List
+import scala.reflect.io.File
+
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.{DataFrame, SparkListenerExtensionTest, SparkSession, SQLContext}
+import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat, CatalogTable, CatalogTableType}
+import org.apache.spark.sql.sources.{BaseRelation, InsertableRelation, SchemaRelationProvider}
+import org.apache.spark.sql.types.{IntegerType, StringType, StructType}
+
+import org.apache.kyuubi.KyuubiFunSuite
+import org.apache.kyuubi.plugin.lineage.events.Lineage
+import org.apache.kyuubi.plugin.lineage.helper.SparkListenerHelper.isSparkVersionAtMost
+
+class SparkSQLLineageParserHelperSuite extends KyuubiFunSuite
+ with SparkListenerExtensionTest {
+
+ val catalogName =
+ if (isSparkVersionAtMost("3.1")) "org.apache.spark.sql.connector.InMemoryTableCatalog"
+ else "org.apache.spark.sql.connector.catalog.InMemoryTableCatalog"
+
+ override protected val catalogImpl: String = "hive"
+
+ override def sparkConf(): SparkConf = {
+ super.sparkConf()
+ .set(
+ "spark.sql.catalog.v2_catalog",
+ catalogName)
+ }
+
+ override def beforeAll(): Unit = {
+ super.beforeAll()
+ spark.sql(s"create database if not exists test_db")
+ spark.sql(s"create database if not exists test_db0")
+ spark.sql(s"create table if not exists test_db0.test_table0" +
Review Comment:
```suggestion
spark.sql("create database if not exists test_db")
spark.sql("create database if not exists test_db0")
spark.sql("create table if not exists test_db0.test_table0" +
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] yaooqinn commented on pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
yaooqinn commented on PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#issuecomment-1225243438
thanks @iodone for the great work, merged to master
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] ulysses-you commented on a diff in pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
ulysses-you commented on code in PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#discussion_r952095219
##########
docs/extensions/engines/spark/lineage.md:
##########
@@ -0,0 +1,171 @@
+<!--
+ - Licensed to the Apache Software Foundation (ASF) under one or more
+ - contributor license agreements. See the NOTICE file distributed with
+ - this work for additional information regarding copyright ownership.
+ - The ASF licenses this file to You under the Apache License, Version 2.0
+ - (the "License"); you may not use this file except in compliance with
+ - the License. You may obtain a copy of the License at
+ -
+ - http://www.apache.org/licenses/LICENSE-2.0
+ -
+ - Unless required by applicable law or agreed to in writing, software
+ - distributed under the License is distributed on an "AS IS" BASIS,
+ - WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ - See the License for the specific language governing permissions and
+ - limitations under the License.
+ -->
+
+# SQL Lineage Support
+
+When we execute a SQL statement, the computing engine will return to a result set,
+each column in the result set may originate from different tables, and these different
+tables depend on other tables, which may have been calculated by functions, aggregation, etc.
+The source table is related to the result set, which is the SQL Lineage.
+
+## Introduction
+
+The current lineage parsing functionality is implemented as a plugin by extending Spark's `QueryExecutionListener`.
+1. The `SparkListenerSQLExecutionEnd` event is triggered after the SQL execution is finished and captured by the `QueryExecuctionListener`,
+ where the SQL lineage parsing process is performed on the successfully executed SQL.
+2. Will write the parsed lineage information to the log file in JSON format.
+
+### Example
+
+When the following SQL is executed:
+```sql
+select a as col0, b as col1 from test_table0
+```
+The lineage of this SQL:
+```json
+{
+ "inputTables":[
+ "default.test_table0"
+ ],
+ "outputTables":[
+
+ ],
+ "columnLineage":[
+ [
+ "col0",
+ [
+ "default.test_table0.a"
+ ]
+ ],
+ [
+ "col1",
+ [
+ "default.test_table0.b"
+ ]
+ ]
+ ]
+}
+```
+
+### SQL type support
+
+Currently supported column lineage for spark's `Command` and `Query` type:
+
+#### Query
+- `Select`
+
+#### Command
+- `CreateDataSourceTableAsSelectCommand`
+- `CreateHiveTableAsSelectCommand`
+- `OptimizedCreateHiveTableAsSelectCommand`
+- `CreateTableAsSelect`
+- `ReplaceTableAsSelect`
+- `InsertIntoDataSourceCommand`
+- `InsertIntoHadoopFsRelationCommand`
+- `InsertIntoDataSourceDirCommand`
+- `InsertIntoHiveDirCommand`
+- `InsertIntoHiveTable`
+
+
+## Building
+
+### Build with Apache Maven
+
+Kyuubi Spark Lineage Listener Extension is built using [Apache Maven](http://maven.apache.org).
+To build it, `cd` to the root direct of kyuubi project and run:
+
+```shell
+build/mvn clean package -pl :kyuubi-spark-lineage_2.12 -DskipTests
+```
+
+After a while, if everything goes well, you will get the plugin finally in two parts:
+
+- The main plugin jar, which is under `./extensions/spark/kyuubi-spark-lineage/target/kyuubi-spark-lineage_${scala.binary.version}-${project.version}.jar`
+- The least transitive dependencies needed, which are under `./extensions/spark/kyuubi-spark-lineage/target/scala-${scala.binary.version}/jars`
+
+### Build against Different Apache Spark Versions
+
+The maven option `spark.version` is used for specifying Spark version to compile with and generate corresponding transitive dependencies.
+By default, it is always built with the latest `spark.version` defined in kyuubi project main pom file.
+Sometimes, it may be incompatible with other Spark distributions, then you may need to build the plugin on your own targeting the Spark version you use.
+
+For example,
+
+```shell
+build/mvn clean package -pl :kyuubi-spark-lineage_2.12 -DskipTests -Dspark.version=3.1.2
+```
+
+The available `spark.version`s are shown in the following table.
+
+| Spark Version | Supported | Remark |
+|:-----------------:|:-----------:|:--------------------------------------------------------------------------------------------------------------------------------:|
+| master | √ | - |
+| 3.3.x | √ | - |
+| 3.2.x | √ | - |
+| 3.1.x | √ | - |
+| 3.0.x | √ | - |
+| 2.4.x | x | - |
+
+Currently, Spark released with Scala 2.12 are supported.
+
+
+### Test with ScalaTest Maven plugin
+If you omit `-DskipTests` option in the command above, you will also get all unit tests run.
+
+```shell
+build/mvn clean package -pl :kyuubi-spark-lineage_2.12
+```
+
+If any bug occurs and you want to debug the plugin yourself, you can configure `-DdebugForkedProcess=true` and `-DdebuggerPort=5005`(optional).
+
+```shell
+build/mvn clean package -pl :kyuubi-spark-lineage_2.12 -DdebugForkedProcess=true
+```
+
+The tests will suspend at startup and wait for a remote debugger to attach to the configured port.
+
+We will appreciate if you can share the bug or the fix to the Kyuubi community.
+
+
+
+## Installing
+
+With the `kyuubi-spark-lineage_*.jar` and its transitive dependencies available for spark runtime classpath, such as
+- Copied to `$SPARK_HOME/jars`, or
+- Specified to `spark.jars` configuration
+
+## Configure
+
+### Settings for Spark Listener Extensions
+
+Add `org.apache.kyuubi.plugin.lineage.SparkOperationLineageQueryExecutionListener` to the spark configuration `spark.sql.queryExecutionListeners`.
+
+```properties
+spark.sql.queryExecutionListeners=org.apache.kyuubi.plugin.lineage.SparkOperationLineageQueryExecutionListener
+```
+
+### Settings for Lineage Logger Path
+
+#### Lineage Logger Path
+The location of all the engine operation lineage events go for the builtin JSON logger.
+All operation lineage events will be written in the unified event json logger path, which be setting with
+`kyuubi.engine.event.json.log.path`. We can get the lineage logger from the `operation_lineage` dir in the
+`kyuubi.engine.event.json.log.path`.
Review Comment:
outdate
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] ulysses-you commented on pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
ulysses-you commented on PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#issuecomment-1223454212
@iodone can you create a issue that support track lineage from subquery and v2write ? we can improve that in future
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] ulysses-you commented on a diff in pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
ulysses-you commented on code in PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#discussion_r952102560
##########
docs/extensions/engines/spark/lineage.md:
##########
@@ -0,0 +1,171 @@
+<!--
+ - Licensed to the Apache Software Foundation (ASF) under one or more
+ - contributor license agreements. See the NOTICE file distributed with
+ - this work for additional information regarding copyright ownership.
+ - The ASF licenses this file to You under the Apache License, Version 2.0
+ - (the "License"); you may not use this file except in compliance with
+ - the License. You may obtain a copy of the License at
+ -
+ - http://www.apache.org/licenses/LICENSE-2.0
+ -
+ - Unless required by applicable law or agreed to in writing, software
+ - distributed under the License is distributed on an "AS IS" BASIS,
+ - WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ - See the License for the specific language governing permissions and
+ - limitations under the License.
+ -->
+
+# SQL Lineage Support
+
+When we execute a SQL statement, the computing engine will return to a result set,
+each column in the result set may originate from different tables, and these different
+tables depend on other tables, which may have been calculated by functions, aggregation, etc.
+The source table is related to the result set, which is the SQL Lineage.
+
+## Introduction
+
+The current lineage parsing functionality is implemented as a plugin by extending Spark's `QueryExecutionListener`.
+1. The `SparkListenerSQLExecutionEnd` event is triggered after the SQL execution is finished and captured by the `QueryExecuctionListener`,
+ where the SQL lineage parsing process is performed on the successfully executed SQL.
+2. Will write the parsed lineage information to the log file in JSON format.
+
+### Example
+
+When the following SQL is executed:
+```sql
+select a as col0, b as col1 from test_table0
+```
+The lineage of this SQL:
+```json
+{
+ "inputTables":[
+ "default.test_table0"
+ ],
+ "outputTables":[
+
+ ],
+ "columnLineage":[
+ [
+ "col0",
+ [
+ "default.test_table0.a"
+ ]
+ ],
+ [
+ "col1",
+ [
+ "default.test_table0.b"
+ ]
+ ]
+ ]
Review Comment:
this structure is not friendly for end user who does not use scala. how about:
```
"columnLineage":[
"col0": [
"default.test_table0.a"
],
"col1": [
"default.test_table0.b"
]
]
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] ulysses-you commented on pull request #3185: [KYUUBI #2282] Add lineage records for sql statement execution in Kyuubi engine logs
Posted by GitBox <gi...@apache.org>.
ulysses-you commented on PR #3185:
URL: https://github.com/apache/incubator-kyuubi/pull/3185#issuecomment-1212881187
yeah, +1 move to extension
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org