Skip to content

Spark 3.5: Add row lineage rule for DELETE to keep plans resolved #18159

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
rishi-rana wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Spark 3.5: Add row lineage rule for DELETE to keep plans resolved #18159

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions ...ns/src/main/scala/org/apache/iceberg/spark/extensions/IcebergSparkSessionExtensions.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ import org.apache.spark.sql.catalyst.analysis.CheckViews
import org.apache.spark.sql.catalyst.analysis.ProcedureArgumentCoercion
import org.apache.spark.sql.catalyst.analysis.ResolveProcedures
import org.apache.spark.sql.catalyst.analysis.ResolveViews
import org.apache.spark.sql.catalyst.analysis.RewriteDeleteFromTableForRowLineage
import org.apache.spark.sql.catalyst.analysis.RewriteMergeIntoTableForRowLineage
import org.apache.spark.sql.catalyst.analysis.RewriteUpdateTableForRowLineage
import org.apache.spark.sql.catalyst.optimizer.RemoveRowLineageOutputFromOriginalTable
Expand All @@ -41,6 +42,7 @@ class IcebergSparkSessionExtensions extends (SparkSessionExtensions => Unit) {
extensions.injectResolutionRule { spark => ResolveViews(spark) }
extensions.injectResolutionRule { _ => ProcedureArgumentCoercion }
extensions.injectCheckRule(_ => CheckViews)
extensions.injectResolutionRule { _ => RewriteDeleteFromTableForRowLineage }
extensions.injectResolutionRule { _ => RewriteUpdateTableForRowLineage }
extensions.injectResolutionRule { _ => RewriteMergeIntoTableForRowLineage }

Expand Down
57 changes: 57 additions & 0 deletions ...in/scala/org/apache/spark/sql/catalyst/analysis/RewriteDeleteFromTableForRowLineage.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,57 @@
/*
* 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.catalyst.analysis

import org.apache.spark.sql.catalyst.plans.logical.DeleteFromTable
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
import org.apache.spark.sql.connector.catalog.SupportsRowLevelOperations
import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation

/**
* Adds the row lineage attributes to the output of a DELETE target so that row lineage is carried
* over to rewritten files. Copy-on-write DELETE replaces whole groups of rows, so surviving rows
* keep the lineage values they already have and no assignments are needed.
*/
object RewriteDeleteFromTableForRowLineage extends RewriteOperationForRowLineage {

override def apply(plan: LogicalPlan): LogicalPlan = {
plan resolveOperators {
case deleteFromTable @ DeleteFromTable(table, _)
if deleteFromTable.resolved && shouldUpdatePlan(table) =>
updatePlanWithRowLineage(deleteFromTable)
}
}

private def updatePlanWithRowLineage(deleteFromTable: DeleteFromTable): LogicalPlan = {
EliminateSubqueryAliases(deleteFromTable.table) match {
case r @ DataSourceV2Relation(_: SupportsRowLevelOperations, _, _, _, _) =>
findRowLineageAttributes(r.metadataOutput) match {
case Some((rowId, lastUpdatedSequence)) =>
val tableWithLineage = r.copy(output = r.output ++ Seq(rowId, lastUpdatedSequence))
deleteFromTable.copy(table = tableWithLineage)

case None =>
deleteFromTable
}

case _ =>
deleteFromTable
}
}
}
1 change: 1 addition & 0 deletions ...src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteOperationForRowLineage.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,7 @@ trait RewriteOperationForRowLineage extends RewriteRowLevelCommand {
r.table match {
case sparkTable: SparkTable =>
TableUtil.supportsRowLineage(sparkTable.table())
case _ => false
}
case _ => false
}
Expand Down
1 change: 1 addition & 0 deletions ...park-extensions/src/test/java/org/apache/iceberg/spark/extensions/ExtensionsTestBase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,7 @@ public static void startMetastoreAndSpark() {
.master("local[2]")
.config("spark.driver.host", InetAddress.getLoopbackAddress().getHostAddress())
.config("spark.testing", "true")
.config("spark.sql.planChangeValidation", "true")
.config(SQLConf.PARTITION_OVERWRITE_MODE().key(), "dynamic")
.config("spark.sql.extensions", IcebergSparkSessionExtensions.class.getName())
.config("spark.hadoop." + METASTOREURIS.varname, hiveConf.get(METASTOREURIS.varname))
Expand Down
Loading