[hudi] branch master updated: Fix race condition in HoodieSparkSqlWriter (#9749)

[vbalaji]

This is an automated email from the ASF dual-hosted git repository.

vbalaji 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 2b245aa7098 Fix race condition in HoodieSparkSqlWriter (#9749)
2b245aa7098 is described below

commit 2b245aa709852f8023a703bd4d574b0491f2c96b
Author: StreamingFlames <18889897...@163.com>
AuthorDate: Sun Oct 22 05:24:16 2023 -0500

    Fix race condition in HoodieSparkSqlWriter (#9749)
---
 .../org/apache/hudi/HoodieSparkSqlWriter.scala     | 47 ++++++++++++++++++++--
 .../TestSparkDataSourceDAGExecution.scala          |  2 +-
 2 files changed, 44 insertions(+), 5 deletions(-)

diff --git 
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala
 
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala
index 581cf2bdc69..49ee4c4b670 100644
--- 
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala
+++ 
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala
@@ -27,6 +27,7 @@ import 
org.apache.hudi.DataSourceOptionsHelper.fetchMissingWriteConfigsFromTable
 import 
org.apache.hudi.DataSourceUtils.tryOverrideParquetWriteLegacyFormatProperty
 import org.apache.hudi.DataSourceWriteOptions._
 import org.apache.hudi.HoodieConversionUtils.{toProperties, toScalaOption}
+import org.apache.hudi.HoodieSparkSqlWriter.{CANONICALIZE_NULLABLE, 
SQL_MERGE_INTO_WRITES, StreamingWriteParams}
 import org.apache.hudi.HoodieWriterUtils._
 import org.apache.hudi.avro.AvroSchemaUtils.{canProject, 
isCompatibleProjectionOf, isSchemaCompatible, resolveNullableSchema}
 import org.apache.hudi.avro.HoodieAvroUtils
@@ -110,6 +111,48 @@ object HoodieSparkSqlWriter {
    */
   val SPARK_STREAMING_BATCH_ID = "hoodie.internal.spark.streaming.batch.id"
 
+  def write(sqlContext: SQLContext,
+            mode: SaveMode,
+            optParams: Map[String, String],
+            sourceDf: DataFrame,
+            streamingWritesParamsOpt: Option[StreamingWriteParams] = 
Option.empty,
+            hoodieWriteClient: Option[SparkRDDWriteClient[_]] = Option.empty):
+  (Boolean, HOption[String], HOption[String], HOption[String], 
SparkRDDWriteClient[_], HoodieTableConfig) = {
+    new HoodieSparkSqlWriterInternal().write(sqlContext, mode, optParams, 
sourceDf, streamingWritesParamsOpt, hoodieWriteClient)
+  }
+
+  def bootstrap(sqlContext: SQLContext,
+                mode: SaveMode,
+                optParams: Map[String, String],
+                df: DataFrame,
+                hoodieTableConfigOpt: Option[HoodieTableConfig] = Option.empty,
+                streamingWritesParamsOpt: Option[StreamingWriteParams] = 
Option.empty,
+                hoodieWriteClient: Option[SparkRDDWriteClient[_]] = 
Option.empty): Boolean = {
+    new HoodieSparkSqlWriterInternal().bootstrap(sqlContext, mode, optParams, 
df, hoodieTableConfigOpt, streamingWritesParamsOpt, hoodieWriteClient)
+  }
+
+  /**
+   * Deduces writer's schema based on
+   * <ul>
+   * <li>Source's schema</li>
+   * <li>Target table's schema (including Hudi's [[InternalSchema]] 
representation)</li>
+   * </ul>
+   */
+  def deduceWriterSchema(sourceSchema: Schema,
+                         latestTableSchemaOpt: Option[Schema],
+                         internalSchemaOpt: Option[InternalSchema],
+                         opts: Map[String, String]): Schema = {
+    new HoodieSparkSqlWriterInternal().deduceWriterSchema(sourceSchema, 
latestTableSchemaOpt, internalSchemaOpt, opts)
+  }
+
+  def cleanup(): Unit = {
+    Metrics.shutdownAllMetrics()
+  }
+
+}
+
+class HoodieSparkSqlWriterInternal {
+
   private val log = LoggerFactory.getLogger(getClass)
   private var tableExists: Boolean = false
   private var asyncCompactionTriggerFnDefined: Boolean = false
@@ -949,10 +992,6 @@ object HoodieSparkSqlWriter {
     }
   }
 
-  def cleanup() : Unit = {
-    Metrics.shutdownAllMetrics()
-  }
-
   private def handleSaveModes(spark: SparkSession, mode: SaveMode, tablePath: 
Path, tableConfig: HoodieTableConfig, tableName: String,
                               operation: WriteOperationType, fs: FileSystem): 
Unit = {
     if (mode == SaveMode.Append && tableExists) {
diff --git 
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestSparkDataSourceDAGExecution.scala
 
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestSparkDataSourceDAGExecution.scala
index 52e1ae812c9..15b4cda243d 100644
--- 
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestSparkDataSourceDAGExecution.scala
+++ 
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestSparkDataSourceDAGExecution.scala
@@ -89,7 +89,7 @@ class TestSparkDataSourceDAGExecution extends 
HoodieSparkClientTestBase with Sca
   @CsvSource(Array(
     "upsert,org.apache.hudi.client.SparkRDDWriteClient.commit",
     "insert,org.apache.hudi.client.SparkRDDWriteClient.commit",
-    "bulk_insert,org.apache.hudi.HoodieSparkSqlWriter$.bulkInsertAsRow"))
+    
"bulk_insert,org.apache.hudi.HoodieSparkSqlWriterInternal.bulkInsertAsRow"))
   def testWriteOperationDoesNotTriggerRepeatedDAG(operation: String, event: 
String): Unit = {
     // register stage event listeners
     val stageListener = new StageListener(event)