lanyuanxiaoyao/hudi
1
0
Fork 0
Code Activity

[MINOR] Minor changes around Spark 3.3 support (#6231)

Browse Source 
Co-authored-by: Shawn Chang <yxchang@amazon.com>
This commit is contained in:
Shawn Chang
2022-07-28 09:32:34 -07:00
committed by GitHub
parent ea1fbc71ec
commit 70b5cf6dab
7 changed files with 7 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/hudi/SparkAdapter.scala
View File

@@ -162,7 +162,7 @@ trait SparkAdapter extends Serializable {
* Extract condition in [[DeleteFromTable]]
* SPARK-38626 condition is no longer Option in Spark 3.3
*/
def extractCondition(deleteFromTable: Command): Expression
def extractDeleteCondition(deleteFromTable: Command): Expression
/**
* Get parseQuery from ExtendedSqlParser, only for Spark 3.3+

2
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/analysis/HoodieAnalysis.scala
View File

@@ -435,7 +435,7 @@ case class HoodieResolveReferences(sparkSession: SparkSession) extends Rule[Logi
// Resolve Delete Table
case dft @ DeleteFromTable(table, condition)
if sparkAdapter.isHoodieTable(table, sparkSession) && table.resolved =>
val resolveExpression = resolveExpressionFrom(table, None)_
val resolveExpression = resolveExpressionFrom(table, None)(_)
sparkAdapter.resolveDeleteFromTable(dft, resolveExpression)
// Append the meta field to the insert query to walk through the validate for the

3
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/DeleteHoodieTableCommand.scala
View File

@@ -37,8 +37,7 @@ case class DeleteHoodieTableCommand(deleteTable: DeleteFromTable) extends Hoodie
// Remove meta fields from the data frame
var df = removeMetaFields(Dataset.ofRows(sparkSession, table))
// SPARK-38626 DeleteFromTable.condition is changed from Option[Expression] to Expression in Spark 3.3
val condition = sparkAdapter.extractCondition(deleteTable)
val condition = sparkAdapter.extractDeleteCondition(deleteTable)
if (condition != null) df = df.filter(Column(condition))
val hoodieCatalogTable = HoodieCatalogTable(sparkSession, tableId)

2
hudi-spark-datasource/hudi-spark2/src/main/scala/org/apache/spark/sql/adapter/Spark2Adapter.scala
View File

@@ -138,7 +138,7 @@ class Spark2Adapter extends SparkAdapter {
DeleteFromTable(deleteFromTableCommand.table, resolvedCondition)
}
override def extractCondition(deleteFromTable: Command): Expression = {
override def extractDeleteCondition(deleteFromTable: Command): Expression = {
deleteFromTable.asInstanceOf[DeleteFromTable].condition.getOrElse(null)
}

2
hudi-spark-datasource/hudi-spark3.1.x/src/main/scala/org/apache/spark/sql/adapter/Spark3_1Adapter.scala
View File

@@ -73,7 +73,7 @@ class Spark3_1Adapter extends BaseSpark3Adapter {
DeleteFromTable(deleteFromTableCommand.table, resolvedCondition)
}
override def extractCondition(deleteFromTable: Command): Expression = {
override def extractDeleteCondition(deleteFromTable: Command): Expression = {
deleteFromTable.asInstanceOf[DeleteFromTable].condition.getOrElse(null)
}
}

2
hudi-spark-datasource/hudi-spark3.2.x/src/main/scala/org/apache/spark/sql/adapter/Spark3_2Adapter.scala
View File

@@ -70,7 +70,7 @@ class Spark3_2Adapter extends BaseSpark3Adapter {
DeleteFromTable(deleteFromTableCommand.table, resolvedCondition)
}
override def extractCondition(deleteFromTable: Command): Expression = {
override def extractDeleteCondition(deleteFromTable: Command): Expression = {
deleteFromTable.asInstanceOf[DeleteFromTable].condition.getOrElse(null)
}
}

2
hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/adapter/Spark3_3Adapter.scala
View File

@@ -69,7 +69,7 @@ class Spark3_3Adapter extends BaseSpark3Adapter {
DeleteFromTable(deleteFromTableCommand.table, resolveExpression(deleteFromTableCommand.condition))
}
override def extractCondition(deleteFromTable: Command): Expression = {
override def extractDeleteCondition(deleteFromTable: Command): Expression = {
deleteFromTable.asInstanceOf[DeleteFromTable].condition
}