lanyuanxiaoyao/hudi
1
0
Fork 0
Code Activity

[HUDI-2520] Fix drop table issue when sync to Hive (#5143)

Browse Source
This commit is contained in:
leesf
2022-03-29 10:34:12 +08:00
committed by GitHub
parent 3bf9c5ffe8
commit 8f8a8158e2
2 changed files with 15 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

0
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/DropHoodieTableCommand.scala → hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/command/DropHoodieTableCommand.scala
View File

29
hudi-spark-datasource/hudi-spark3/src/main/scala/org/apache/spark/sql/hudi/catalog/HoodieCatalog.scala
View File

@@ -19,12 +19,10 @@
package org.apache.spark.sql.hudi.catalog package org.apache.spark.sql.hudi.catalog
import org.apache.hadoop.fs.Path import org.apache.hadoop.fs.Path
import org.apache.hudi.{DataSourceWriteOptions, SparkAdapterSupport}
import org.apache.hudi.client.common.HoodieSparkEngineContext
import org.apache.hudi.common.fs.FSUtils
import org.apache.hudi.exception.HoodieException import org.apache.hudi.exception.HoodieException
import org.apache.hudi.hive.util.ConfigUtils import org.apache.hudi.hive.util.ConfigUtils
import org.apache.hudi.sql.InsertMode import org.apache.hudi.sql.InsertMode
import org.apache.hudi.{DataSourceWriteOptions, SparkAdapterSupport}
import org.apache.spark.sql.HoodieSpark3SqlUtils.convertTransforms import org.apache.spark.sql.HoodieSpark3SqlUtils.convertTransforms
import org.apache.spark.sql.catalyst.TableIdentifier import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.catalyst.analysis.{NoSuchTableException, TableAlreadyExistsException, UnresolvedAttribute} import org.apache.spark.sql.catalyst.analysis.{NoSuchTableException, TableAlreadyExistsException, UnresolvedAttribute}
@@ -34,7 +32,7 @@ import org.apache.spark.sql.connector.catalog.TableChange.{AddColumn, ColumnChan
import org.apache.spark.sql.connector.catalog._ import org.apache.spark.sql.connector.catalog._
import org.apache.spark.sql.connector.expressions.Transform import org.apache.spark.sql.connector.expressions.Transform
import org.apache.spark.sql.execution.datasources.DataSource import org.apache.spark.sql.execution.datasources.DataSource
import org.apache.spark.sql.hudi.command.{AlterHoodieTableAddColumnsCommand, AlterHoodieTableChangeColumnCommand, AlterHoodieTableRenameCommand, CreateHoodieTableCommand} import org.apache.spark.sql.hudi.command._
import org.apache.spark.sql.hudi.{HoodieSqlCommonUtils, ProvidesHoodieConfig} import org.apache.spark.sql.hudi.{HoodieSqlCommonUtils, ProvidesHoodieConfig}
import org.apache.spark.sql.types.{StructField, StructType} import org.apache.spark.sql.types.{StructField, StructType}
import org.apache.spark.sql.{Dataset, SaveMode, SparkSession, _} import org.apache.spark.sql.{Dataset, SaveMode, SparkSession, _}
@@ -116,21 +114,24 @@ class HoodieCatalog extends DelegatingCatalogExtension
override def tableExists(ident: Identifier): Boolean = super.tableExists(ident) override def tableExists(ident: Identifier): Boolean = super.tableExists(ident)
override def dropTable(ident: Identifier): Boolean = super.dropTable(ident) override def dropTable(ident: Identifier): Boolean = {
val table = loadTable(ident)
table match {
case _: HoodieInternalV2Table =>
DropHoodieTableCommand(ident.asTableIdentifier, ifExists = true, isView = false, purge = false).run(spark)
true
case _ => super.dropTable(ident)
}
}
override def purgeTable(ident: Identifier): Boolean = { override def purgeTable(ident: Identifier): Boolean = {
val table = loadTable(ident) val table = loadTable(ident)
table match { table match {
case hoodieTable: HoodieInternalV2Table => case _: HoodieInternalV2Table =>
val location = hoodieTable.hoodieCatalogTable.tableLocation DropHoodieTableCommand(ident.asTableIdentifier, ifExists = true, isView = false, purge = true).run(spark)
val targetPath = new Path(location) true
val engineContext = new HoodieSparkEngineContext(spark.sparkContext) case _ => super.purgeTable(ident)
val fs = FSUtils.getFs(location, spark.sparkContext.hadoopConfiguration)
FSUtils.deleteDir(engineContext, fs, targetPath, spark.sparkContext.defaultParallelism)
super.dropTable(ident)
case _ =>
} }
true
} }
@throws[NoSuchTableException] @throws[NoSuchTableException]