[DO NOT MERGE] [HUDI-6198] Run gh actions with Spark 3.4.0

.github/workflows/bot.yml

@@ -27,20 +27,8 @@ jobs:
     strategy:
       matrix:
         include:
-          - scalaProfile: "scala-2.11"
-            sparkProfile: "spark2.4"
-
-          - scalaProfile: "scala-2.12"
-            sparkProfile: "spark2.4"
-
-          - scalaProfile: "scala-2.12"
-            sparkProfile: "spark3.1"
-
-          - scalaProfile: "scala-2.12"
-            sparkProfile: "spark3.2"
-
           - scalaProfile: "scala-2.12"
-            sparkProfile: "spark3.3"
+            sparkProfile: "spark3.4"
 
     steps:
       - uses: actions/checkout@v2
@@ -75,9 +63,6 @@ jobs:
     strategy:
       matrix:
         include:
-          - flinkProfile: "flink1.13"
-          - flinkProfile: "flink1.14"
-          - flinkProfile: "flink1.15"
           - flinkProfile: "flink1.16"
     steps:
       - uses: actions/checkout@v2
@@ -106,13 +91,7 @@ jobs:
       matrix:
         include:
           - flinkProfile: 'flink1.16'
-            sparkProfile: 'spark3.3'
-          - flinkProfile: 'flink1.15'
-            sparkProfile: 'spark3.3'
-          - flinkProfile: 'flink1.14'
-            sparkProfile: 'spark3.2'
-          - flinkProfile: 'flink1.13'
-            sparkProfile: 'spark3.1'
+            sparkProfile: 'spark3.4'
     steps:
       - uses: actions/checkout@v2
       - name: Set up JDK 8

hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkFileReaderFactory.java

@@ -33,6 +33,8 @@ protected HoodieFileReader newParquetFileReader(Configuration conf, Path path) {
     conf.setIfUnset(SQLConf.PARQUET_INT96_AS_TIMESTAMP().key(),
         SQLConf.PARQUET_INT96_AS_TIMESTAMP().defaultValueString());
     conf.setIfUnset(SQLConf.CASE_SENSITIVE().key(), SQLConf.CASE_SENSITIVE().defaultValueString());
+    conf.setIfUnset(SQLConf.PARQUET_INFER_TIMESTAMP_NTZ_ENABLED().key(), SQLConf.PARQUET_INFER_TIMESTAMP_NTZ_ENABLED().defaultValueString());
+    conf.setIfUnset(SQLConf.LEGACY_PARQUET_NANOS_AS_LONG().key(), SQLConf.LEGACY_PARQUET_NANOS_AS_LONG().defaultValueString());
     return new HoodieSparkParquetReader(conf, path);
   }
 

hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/HoodieSparkUtils.scala

@@ -44,6 +44,7 @@ private[hudi] trait SparkVersionsSupport {
   def isSpark3_1: Boolean = getSparkVersion.startsWith("3.1")
   def isSpark3_2: Boolean = getSparkVersion.startsWith("3.2")
   def isSpark3_3: Boolean = getSparkVersion.startsWith("3.3")
+  def isSpark3_4: Boolean = getSparkVersion.startsWith("3.4")
 
   def gteqSpark3_0: Boolean = getSparkVersion >= "3.0"
   def gteqSpark3_1: Boolean = getSparkVersion >= "3.1"
@@ -52,6 +53,8 @@ private[hudi] trait SparkVersionsSupport {
   def gteqSpark3_2_1: Boolean = getSparkVersion >= "3.2.1"
   def gteqSpark3_2_2: Boolean = getSparkVersion >= "3.2.2"
   def gteqSpark3_3: Boolean = getSparkVersion >= "3.3"
+  def gteqSpark3_3_2: Boolean = getSparkVersion >= "3.3.2"
+  def gteqSpark3_4: Boolean = getSparkVersion >= "3.4"
 }
 
 object HoodieSparkUtils extends SparkAdapterSupport with SparkVersionsSupport {

hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/SparkAdapterSupport.scala

@@ -33,7 +33,9 @@ trait SparkAdapterSupport {
 object SparkAdapterSupport {
 
   lazy val sparkAdapter: SparkAdapter = {
-    val adapterClass =  if (HoodieSparkUtils.isSpark3_3) {
+    val adapterClass =  if (HoodieSparkUtils.isSpark3_4) {
+      "org.apache.spark.sql.adapter.Spark3_4Adapter"
+    } else if (HoodieSparkUtils.isSpark3_3) {
       "org.apache.spark.sql.adapter.Spark3_3Adapter"
     } else if (HoodieSparkUtils.isSpark3_2) {
       "org.apache.spark.sql.adapter.Spark3_2Adapter"

hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/BaseFileOnlyRelation.scala

@@ -200,7 +200,7 @@ case class BaseFileOnlyRelation(override val sqlContext: SQLContext,
           // NOTE: We have to specify table's base-path explicitly, since we're requesting Spark to read it as a
           //       list of globbed paths which complicates partitioning discovery for Spark.
           //       Please check [[PartitioningAwareFileIndex#basePaths]] comment for more details.
-          PartitioningAwareFileIndex.BASE_PATH_PARAM -> metaClient.getBasePathV2.toString
+          FileIndexOptions.BASE_PATH_PARAM -> metaClient.getBasePathV2.toString
         ),
         partitionColumns = partitionColumns
       )

hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieBaseRelation.scala

@@ -549,7 +549,7 @@ abstract class HoodieBaseRelation(val sqlContext: SQLContext,
 
     BaseFileReader(
       read = partitionedFile => {
-        val extension = FSUtils.getFileExtension(partitionedFile.filePath)
+        val extension = FSUtils.getFileExtension(partitionedFile.filePath.toString)
         if (tableBaseFileFormat.getFileExtension.equals(extension)) {
           read(partitionedFile)
         } else {
@@ -688,7 +688,7 @@ object HoodieBaseRelation extends SparkAdapterSupport {
 
     partitionedFile => {
       val hadoopConf = hadoopConfBroadcast.value.get()
-      val reader = new HoodieAvroHFileReader(hadoopConf, new Path(partitionedFile.filePath),
+      val reader = new HoodieAvroHFileReader(hadoopConf, new Path(partitionedFile.filePath.toString),
         new CacheConfig(hadoopConf))
 
       val requiredRowSchema = requiredDataSchema.structTypeSchema

hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieBootstrapRelation.scala

@@ -25,6 +25,7 @@ import org.apache.hudi.common.table.{HoodieTableMetaClient, TableSchemaResolver}
 import org.apache.hudi.exception.HoodieException
 import org.apache.spark.execution.datasources.HoodieInMemoryFileIndex
 import org.apache.spark.internal.Logging
+import org.apache.spark.paths.SparkPath
 import org.apache.spark.rdd.RDD
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.execution.datasources.{FileStatusCache, PartitionedFile}
@@ -79,11 +80,11 @@ class HoodieBootstrapRelation(@transient val _sqlContext: SQLContext,
       var dataFile: PartitionedFile = null
 
       if (hoodieBaseFile.getBootstrapBaseFile.isPresent) {
-        skeletonFile = Option(PartitionedFile(InternalRow.empty, hoodieBaseFile.getPath, 0, hoodieBaseFile.getFileLen))
-        dataFile = PartitionedFile(InternalRow.empty, hoodieBaseFile.getBootstrapBaseFile.get().getPath, 0,
+        skeletonFile = Option(PartitionedFile(InternalRow.empty, SparkPath.fromPathString(hoodieBaseFile.getPath), 0, hoodieBaseFile.getFileLen))
+        dataFile = PartitionedFile(InternalRow.empty, SparkPath.fromPathString(hoodieBaseFile.getBootstrapBaseFile.get().getPath), 0,
           hoodieBaseFile.getBootstrapBaseFile.get().getFileLen)
       } else {
-        dataFile = PartitionedFile(InternalRow.empty, hoodieBaseFile.getPath, 0, hoodieBaseFile.getFileLen)
+        dataFile = PartitionedFile(InternalRow.empty, SparkPath.fromPathString(hoodieBaseFile.getPath), 0, hoodieBaseFile.getFileLen)
       }
       HoodieBootstrapSplit(dataFile, skeletonFile)
     })

hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieDataSourceHelper.scala

@@ -27,6 +27,7 @@ import org.apache.hudi.common.util.StringUtils.isNullOrEmpty
 import org.apache.hudi.common.util.ValidationUtils.checkState
 import org.apache.hudi.internal.schema.InternalSchema
 import org.apache.hudi.internal.schema.utils.SerDeHelper
+import org.apache.spark.paths.SparkPath
 import org.apache.spark.sql.SparkSession
 import org.apache.spark.sql.avro.HoodieAvroDeserializer
 import org.apache.spark.sql.catalyst.InternalRow
@@ -85,7 +86,7 @@ object HoodieDataSourceHelper extends PredicateHelper with SparkAdapterSupport {
     (0L until file.getLen by maxSplitBytes).map { offset =>
       val remaining = file.getLen - offset
       val size = if (remaining > maxSplitBytes) maxSplitBytes else remaining
-      PartitionedFile(partitionValues, filePath.toUri.toString, offset, size)
+      PartitionedFile(partitionValues, SparkPath.fromPath(filePath), offset, size)
     }
   }
 

hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/Iterators.scala

@@ -342,7 +342,7 @@ object LogFileIterator {
     // Determine partition path as an immediate parent folder of either
     //    - The base file
     //    - Some log file
-    split.dataFile.map(baseFile => new Path(baseFile.filePath))
+    split.dataFile.map(baseFile => new Path(baseFile.filePath.toString))
       .getOrElse(split.logFiles.head.getPath)
       .getParent
   }

hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/MergeOnReadSnapshotRelation.scala

@@ -22,10 +22,11 @@ import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.fs.Path
 import org.apache.hudi.HoodieBaseRelation.convertToAvroSchema
 import org.apache.hudi.HoodieConversionUtils.toScalaOption
-import org.apache.hudi.MergeOnReadSnapshotRelation.{getFilePath, isProjectionCompatible}
+import org.apache.hudi.MergeOnReadSnapshotRelation.isProjectionCompatible
 import org.apache.hudi.avro.HoodieAvroUtils
 import org.apache.hudi.common.model.{FileSlice, HoodieLogFile, OverwriteWithLatestAvroPayload}
 import org.apache.hudi.common.table.HoodieTableMetaClient
+import org.apache.spark.paths.SparkPath
 import org.apache.spark.rdd.RDD
 import org.apache.spark.sql.SQLContext
 import org.apache.spark.sql.catalyst.InternalRow
@@ -234,8 +235,7 @@ abstract class BaseMergeOnReadSnapshotRelation(sqlContext: SQLContext,
       val logFiles = fileSlice.getLogFiles.sorted(HoodieLogFile.getLogFileComparator).iterator().asScala.toList
 
       val partitionedBaseFile = baseFile.map { file =>
-        val filePath = getFilePath(file.getFileStatus.getPath)
-        PartitionedFile(getPartitionColumnsAsInternalRow(file.getFileStatus), filePath, 0, file.getFileLen)
+        PartitionedFile(getPartitionColumnsAsInternalRow(file.getFileStatus), SparkPath.fromPath(file.getFileStatus.getPath), 0, file.getFileLen)
       }
 
       HoodieMergeOnReadFileSplit(partitionedBaseFile, logFiles)
@@ -259,21 +259,4 @@ object MergeOnReadSnapshotRelation {
 
   def isProjectionCompatible(tableState: HoodieTableState): Boolean =
     projectionCompatiblePayloadClasses.contains(tableState.recordPayloadClassName)
-
-  def getFilePath(path: Path): String = {
-    // Here we use the Path#toUri to encode the path string, as there is a decode in
-    // ParquetFileFormat#buildReaderWithPartitionValues in the spark project when read the table
-    // .So we should encode the file path here. Otherwise, there is a FileNotException throw
-    // out.
-    // For example, If the "pt" is the partition path field, and "pt" = "2021/02/02", If
-    // we enable the URL_ENCODE_PARTITIONING and write data to hudi table.The data path
-    // in the table will just like "/basePath/2021%2F02%2F02/xxxx.parquet". When we read
-    // data from the table, if there are no encode for the file path,
-    // ParquetFileFormat#buildReaderWithPartitionValues will decode it to
-    // "/basePath/2021/02/02/xxxx.parquet" witch will result to a FileNotException.
-    // See FileSourceScanExec#createBucketedReadRDD in spark project which do the same thing
-    // when create PartitionedFile.
-    path.toUri.toString
-  }
-
 }

hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/cdc/HoodieCDCRDD.scala

@@ -42,6 +42,7 @@ import org.apache.hudi.config.{HoodiePayloadConfig, HoodieWriteConfig}
 import org.apache.hudi.keygen.constant.KeyGeneratorOptions
 import org.apache.hudi.keygen.factory.HoodieSparkKeyGeneratorFactory
 import org.apache.hudi.{AvroConversionUtils, AvroProjection, HoodieFileIndex, HoodieMergeOnReadFileSplit, HoodieTableSchema, HoodieTableState, HoodieUnsafeRDD, LogFileIterator, RecordMergingFileIterator, SparkAdapterSupport}
+import org.apache.spark.paths.SparkPath
 import org.apache.spark.rdd.RDD
 import org.apache.spark.sql.HoodieCatalystExpressionUtils.generateUnsafeProjection
 import org.apache.spark.sql.SparkSession
@@ -419,7 +420,7 @@ class HoodieCDCRDD(
             assert(currentCDCFileSplit.getCdcFiles != null && currentCDCFileSplit.getCdcFiles.size() == 1)
             val absCDCPath = new Path(basePath, currentCDCFileSplit.getCdcFiles.get(0))
             val fileStatus = fs.getFileStatus(absCDCPath)
-            val pf = PartitionedFile(InternalRow.empty, absCDCPath.toUri.toString, 0, fileStatus.getLen)
+            val pf = PartitionedFile(InternalRow.empty, SparkPath.fromPath(absCDCPath), 0, fileStatus.getLen)
             recordIter = parquetReader(pf)
           case BASE_FILE_DELETE =>
             assert(currentCDCFileSplit.getBeforeFileSlice.isPresent)
@@ -525,7 +526,7 @@ class HoodieCDCRDD(
       val baseFileStatus = fs.getFileStatus(new Path(fileSlice.getBaseFile.get().getPath))
       val basePartitionedFile = PartitionedFile(
         InternalRow.empty,
-        pathToString(baseFileStatus.getPath),
+        SparkPath.fromPath(baseFileStatus.getPath),
         0,
         baseFileStatus.getLen
       )

hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/command/AlterHoodieTableAddColumnsCommand.scala

@@ -84,7 +84,6 @@ case class AlterHoodieTableAddColumnsCommand(
 
     SchemaUtils.checkColumnNameDuplication(
       newSqlDataSchema.map(_.name),
-      "in the table definition of " + table.identifier,
       conf.caseSensitiveAnalysis)
 
     sparkSession.sessionState.catalog.alterTableDataSchema(tableId, newSqlDataSchema)

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

@@ -62,7 +62,9 @@ object HoodieAnalysis {
         session => instantiateKlass(spark3AnalysisClass, session)
 
       val resolveAlterTableCommandsClass =
-        if (HoodieSparkUtils.gteqSpark3_3)
+        if (HoodieSparkUtils.gteqSpark3_4)
+          "org.apache.spark.sql.hudi.Spark34ResolveHudiAlterTableCommand"
+        else if (HoodieSparkUtils.gteqSpark3_3)
           "org.apache.spark.sql.hudi.Spark33ResolveHudiAlterTableCommand"
         else "org.apache.spark.sql.hudi.Spark32ResolveHudiAlterTableCommand"
       val resolveAlterTableCommands: RuleBuilder =
@@ -106,7 +108,9 @@ object HoodieAnalysis {
     val optimizerRules = ListBuffer[RuleBuilder]()
     if (HoodieSparkUtils.gteqSpark3_1) {
       val nestedSchemaPruningClass =
-        if (HoodieSparkUtils.gteqSpark3_3) {
+        if (HoodieSparkUtils.gteqSpark3_4)
+          "org.apache.spark.sql.execution.datasources.Spark34NestedSchemaPruning"
+        else if (HoodieSparkUtils.gteqSpark3_3) {
           "org.apache.spark.sql.execution.datasources.Spark33NestedSchemaPruning"
         } else if (HoodieSparkUtils.gteqSpark3_2) {
           "org.apache.spark.sql.execution.datasources.Spark32NestedSchemaPruning"
@@ -160,7 +164,7 @@ case class HoodieAnalysis(sparkSession: SparkSession) extends Rule[LogicalPlan]
   override def apply(plan: LogicalPlan): LogicalPlan = {
     plan match {
       // Convert to MergeIntoHoodieTableCommand
-      case m @ MergeIntoTable(target, _, _, _, _)
+      case m @ MergeIntoTable(target, _, _, _, _,_)
         if m.resolved && sparkAdapter.isHoodieTable(target, sparkSession) =>
           MergeIntoHoodieTableCommand(m)
 
@@ -287,7 +291,7 @@ case class HoodieResolveReferences(sparkSession: SparkSession) extends Rule[Logi
 
   def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperatorsUp {
     // Resolve merge into
-    case mergeInto @ MergeIntoTable(target, source, mergeCondition, matchedActions, notMatchedActions)
+    case mergeInto @ MergeIntoTable(target, source, mergeCondition, matchedActions, notMatchedActions, _)
       if sparkAdapter.isHoodieTable(target, sparkSession) && target.resolved =>
       val resolver = sparkSession.sessionState.conf.resolver
       val resolvedSource = analyzer.execute(source)
@@ -455,7 +459,7 @@ case class HoodieResolveReferences(sparkSession: SparkSession) extends Rule[Logi
       }
       // Return the resolved MergeIntoTable
       MergeIntoTable(target, resolvedSource, resolvedMergeCondition,
-        resolvedMatchedActions, resolvedNotMatchedActions)
+        resolvedMatchedActions, resolvedNotMatchedActions, Seq.empty)
 
     // Resolve update table
     case UpdateTable(table, assignments, condition)

hudi-spark-datasource/hudi-spark3.2plus-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark32PlusHoodieParquetFileFormat.scala

@@ -126,7 +126,7 @@ class Spark32PlusHoodieParquetFileFormat(private val shouldAppendPartitionValues
     val pushDownDate = sqlConf.parquetFilterPushDownDate
     val pushDownTimestamp = sqlConf.parquetFilterPushDownTimestamp
     val pushDownDecimal = sqlConf.parquetFilterPushDownDecimal
-    val pushDownStringStartWith = sqlConf.parquetFilterPushDownStringStartWith
+    val pushDownStringStartWith = sqlConf.parquetFilterPushDownStringPredicate
     val pushDownInFilterThreshold = sqlConf.parquetFilterPushDownInFilterThreshold
     val isCaseSensitive = sqlConf.caseSensitiveAnalysis
     val parquetOptions = new ParquetOptions(options, sparkSession.sessionState.conf)
@@ -136,7 +136,7 @@ class Spark32PlusHoodieParquetFileFormat(private val shouldAppendPartitionValues
     (file: PartitionedFile) => {
       assert(!shouldAppendPartitionValues || file.partitionValues.numFields == partitionSchema.size)
 
-      val filePath = new Path(new URI(file.filePath))
+      val filePath = file.filePath.toPath
       val split = new FileSplit(filePath, file.start, file.length, Array.empty[String])
 
       val sharedConf = broadcastedHadoopConf.value.value