You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by da...@apache.org on 2024/06/16 00:15:59 UTC
(hudi) branch master updated: [HUDI-7880] Support extraMetadata in Spark SQL Insert Into (#11458)
This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new ec5244d5fe6 [HUDI-7880] Support extraMetadata in Spark SQL Insert Into (#11458)
ec5244d5fe6 is described below
commit ec5244d5fe6a3b376dbfc4d294f5e139c6ce1e84
Author: 董可伦 <do...@inspur.com>
AuthorDate: Sun Jun 16 08:15:48 2024 +0800
[HUDI-7880] Support extraMetadata in Spark SQL Insert Into (#11458)
---
.../scala/org/apache/hudi/HoodieCLIUtils.scala | 4 +-
.../spark/sql/hudi/HoodieSqlCommonUtils.scala | 10 +++-
.../spark/sql/hudi/ProvidesHoodieConfig.scala | 5 +-
.../spark/sql/hudi/dml/TestInsertTable.scala | 53 +++++++++++++++++++++-
4 files changed, 63 insertions(+), 9 deletions(-)
diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieCLIUtils.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieCLIUtils.scala
index 03e5f2820a3..10079071594 100644
--- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieCLIUtils.scala
+++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieCLIUtils.scala
@@ -32,7 +32,7 @@ import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.catalyst.catalog.HoodieCatalogTable
import org.apache.spark.sql.hudi.HoodieOptionConfig
-import org.apache.spark.sql.hudi.HoodieSqlCommonUtils.isHoodieConfigKey
+import org.apache.spark.sql.hudi.HoodieSqlCommonUtils.filterHoodieConfigs
import scala.collection.JavaConverters.{collectionAsScalaIterableConverter, mapAsJavaMapConverter, propertiesAsScalaMapConverter}
@@ -58,7 +58,7 @@ object HoodieCLIUtils {
val finalParameters = HoodieWriterUtils.parametersWithWriteDefaults(
(catalogProps ++
metaClient.getTableConfig.getProps.asScala.toMap ++
- sparkSession.sqlContext.getAllConfs.filterKeys(isHoodieConfigKey) ++
+ filterHoodieConfigs(sparkSession.sqlContext.getAllConfs) ++
conf).toMap
)
diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/HoodieSqlCommonUtils.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/HoodieSqlCommonUtils.scala
index dc2f5c85f04..719327cc42c 100644
--- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/HoodieSqlCommonUtils.scala
+++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/HoodieSqlCommonUtils.scala
@@ -17,6 +17,7 @@
package org.apache.spark.sql.hudi
+import org.apache.hudi.DataSourceWriteOptions.COMMIT_METADATA_KEYPREFIX
import org.apache.hudi.client.common.HoodieSparkEngineContext
import org.apache.hudi.common.config.{HoodieMetadataConfig, TypedProperties}
import org.apache.hudi.common.fs.FSUtils
@@ -226,8 +227,13 @@ object HoodieSqlCommonUtils extends SparkAdapterSupport {
*
* TODO: standardize the key prefix so that we don't need this helper (HUDI-4935)
*/
- def isHoodieConfigKey(key: String): Boolean =
- key.startsWith("hoodie.") || key == DataSourceReadOptions.TIME_TRAVEL_AS_OF_INSTANT.key
+ private def isHoodieConfigKey(key: String, commitMetadataKeyPrefix: String): Boolean =
+ key.startsWith("hoodie.") || key.startsWith(commitMetadataKeyPrefix) ||
+ key == DataSourceReadOptions.TIME_TRAVEL_AS_OF_INSTANT.key
+
+ def filterHoodieConfigs(opts: Map[String, String]): Map[String, String] =
+ opts.filterKeys(isHoodieConfigKey(_,
+ opts.getOrElse(COMMIT_METADATA_KEYPREFIX.key, COMMIT_METADATA_KEYPREFIX.defaultValue()))).toMap
/**
* Checks whether Spark is using Hive as Session's Catalog
diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/ProvidesHoodieConfig.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/ProvidesHoodieConfig.scala
index 0af2e59e23c..718c6c08dc5 100644
--- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/ProvidesHoodieConfig.scala
+++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/ProvidesHoodieConfig.scala
@@ -38,7 +38,7 @@ import org.apache.spark.sql.catalyst.expressions.{AttributeReference, EqualTo, L
import org.apache.spark.sql.execution.datasources.FileStatusCache
import org.apache.spark.sql.hive.HiveExternalCatalog
import org.apache.spark.sql.hudi.HoodieOptionConfig.mapSqlOptionsToDataSourceWriteConfigs
-import org.apache.spark.sql.hudi.HoodieSqlCommonUtils.{isHoodieConfigKey, isUsingHiveCatalog}
+import org.apache.spark.sql.hudi.HoodieSqlCommonUtils.{filterHoodieConfigs, isUsingHiveCatalog}
import org.apache.spark.sql.hudi.ProvidesHoodieConfig.{combineOptions, getPartitionPathFieldWriteConfig}
import org.apache.spark.sql.hudi.command.{SqlKeyGenerator, ValidateDuplicateKeyPayload}
import org.apache.spark.sql.internal.SQLConf
@@ -575,7 +575,4 @@ object ProvidesHoodieConfig {
private def filterNullValues(opts: Map[String, String]): Map[String, String] =
opts.filter { case (_, v) => v != null }
- private def filterHoodieConfigs(opts: Map[String, String]): Map[String, String] =
- opts.filterKeys(isHoodieConfigKey).toMap
-
}
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/TestInsertTable.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/TestInsertTable.scala
index 66a5a2ab469..3994fc1bcca 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/TestInsertTable.scala
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/TestInsertTable.scala
@@ -19,7 +19,6 @@ package org.apache.spark.sql.hudi.dml
import org.apache.hudi.DataSourceWriteOptions._
import org.apache.hudi.client.common.HoodieSparkEngineContext
-import org.apache.hudi.common.config.HoodieMetadataConfig
import org.apache.hudi.common.model.HoodieRecord.HoodieRecordType
import org.apache.hudi.common.model.{HoodieRecord, WriteOperationType}
import org.apache.hudi.common.table.{HoodieTableConfig, TableSchemaResolver}
@@ -2585,6 +2584,58 @@ class TestInsertTable extends HoodieSparkSqlTestBase {
})
}
+ test("Test Insert Into with extraMetadata") {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = s"${tmp.getCanonicalPath}/$tableName"
+
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | dt string,
+ | name string,
+ | price double,
+ | ts long
+ |) using hudi
+ | tblproperties (primaryKey = 'id')
+ | partitioned by (dt)
+ | location '$tablePath'
+ """.stripMargin)
+
+ spark.sql("set hoodie.datasource.write.commitmeta.key.prefix=commit_extra_meta_")
+
+ spark.sql("set commit_extra_meta_a=valA")
+ spark.sql("set commit_extra_meta_b=valB")
+ spark.sql(s"insert into $tableName values (1, 'a1', 10, 1000, '2024-06-14')")
+
+ assertResult("valA") {
+ getLastCommitMetadata(spark, tablePath).getExtraMetadata.get("commit_extra_meta_a")
+ }
+ assertResult("valB") {
+ getLastCommitMetadata(spark, tablePath).getExtraMetadata.get("commit_extra_meta_b")
+ }
+ checkAnswer(s"select id, name, price, dt from $tableName")(
+ Seq(1, "a1", 10.0, "2024-06-14")
+ )
+
+ spark.sql("set commit_extra_meta_a=new_valA")
+ spark.sql("set commit_extra_meta_b=new_valB")
+ spark.sql(s"insert into $tableName values (2, 'a2', 20, 2000, '2024-06-14')")
+
+ assertResult("new_valA") {
+ getLastCommitMetadata(spark, tablePath).getExtraMetadata.get("commit_extra_meta_a")
+ }
+ assertResult("new_valB") {
+ getLastCommitMetadata(spark, tablePath).getExtraMetadata.get("commit_extra_meta_b")
+ }
+ checkAnswer(s"select id, name, price, dt from $tableName")(
+ Seq(1, "a1", 10.0, "2024-06-14"),
+ Seq(2, "a2", 20.0, "2024-06-14")
+ )
+ }
+ }
+
def ingestAndValidateDataDupPolicy(tableType: String, tableName: String, tmp: File,
expectedOperationtype: WriteOperationType = WriteOperationType.INSERT,
setOptions: List[String] = List.empty,