Merge branch 'fix-clean-command' into 'release-0.11.0' (merge request !56)

fix RunCleanProcedure's ArrayIndexOutOfBoundsException

Author: XuQianJin-Stars

hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/RunCleanProcedure.scala

@@ -17,6 +17,7 @@
 
 package org.apache.spark.sql.hudi.command.procedures
 
+import java.util.function.Supplier
 import org.apache.hudi.HoodieCLIUtils
 import org.apache.hudi.common.table.timeline.HoodieActiveTimeline
 import org.apache.hudi.common.util.JsonUtils
@@ -25,7 +26,7 @@ import org.apache.spark.internal.Logging
 import org.apache.spark.sql.Row
 import org.apache.spark.sql.types.{DataTypes, Metadata, StructField, StructType}
 
-import java.util.function.Supplier
+import java.util
 
 class RunCleanProcedure extends BaseProcedure with ProcedureBuilder with Logging {
 
@@ -74,16 +75,20 @@ class RunCleanProcedure extends BaseProcedure with ProcedureBuilder with Logging
     if (cleanPolicy.isDefined) {
       props += (HoodieCleanConfig.CLEANER_POLICY.key() -> String.valueOf(cleanPolicy.get))
     }
-    val client = HoodieCLIUtils.createHoodieClientFromPath(sparkSession, basePath, props, tableName)
+    val client = HoodieCLIUtils.createHoodieClientFromPath(sparkSession, basePath, props)
     val hoodieCleanMeta = client.clean(cleanInstantTime, scheduleInLine, skipLocking)
+    val rows = new util.ArrayList[Row]
+
+    if (hoodieCleanMeta != null) {
+      rows.add(Row(hoodieCleanMeta.getStartCleanTime,
+        hoodieCleanMeta.getTimeTakenInMillis,
+        hoodieCleanMeta.getTotalFilesDeleted,
+        hoodieCleanMeta.getEarliestCommitToRetain,
+        JsonUtils.getObjectMapper.writeValueAsString(hoodieCleanMeta.getBootstrapPartitionMetadata),
+        hoodieCleanMeta.getVersion))
+    }
 
-    if (hoodieCleanMeta == null) Seq(Row.empty)
-    else Seq(Row(hoodieCleanMeta.getStartCleanTime,
-      hoodieCleanMeta.getTimeTakenInMillis,
-      hoodieCleanMeta.getTotalFilesDeleted,
-      hoodieCleanMeta.getEarliestCommitToRetain,
-      JsonUtils.getObjectMapper.writeValueAsString(hoodieCleanMeta.getBootstrapPartitionMetadata),
-      hoodieCleanMeta.getVersion))
+    rows.stream().toArray().map(r => r.asInstanceOf[Row]).toList
   }
 }