[Spark] fix auto compaction task bug when partition are multi levels (l…

…akesoul-io#516) Signed-off-by: fphantam <[email protected]>
zjdMASTER · Jul 24, 2024 · ec7fe88 · ec7fe88
1 parent b481925
commit ec7fe88
Showing 1 changed file with 7 additions and 3 deletions.
diff --git a/lakesoul-spark/src/main/scala/com/dmetasoul/lakesoul/spark/compaction/CompactionTask.scala b/lakesoul-spark/src/main/scala/com/dmetasoul/lakesoul/spark/compaction/CompactionTask.scala
@@ -78,8 +78,7 @@ object CompactionTask {
               val partitionDesc = jsonObj.get("table_partition_desc").getAsString
               val tableNamespace = jsonObj.get("table_namespace").getAsString
               if (tableNamespace.equals(database) || database.equals("")) {
-                val rsPartitionDesc = if (partitionDesc.equals(MetaUtils.DEFAULT_RANGE_PARTITION_VALUE)) "" else partitionDesc.replace("=",
-                  "='") + "'"
+                val rsPartitionDesc = if (partitionDesc.equals(MetaUtils.DEFAULT_RANGE_PARTITION_VALUE)) "" else partitionDesc
                 threadPool.execute(new CompactionTableInfo(tablePath, rsPartitionDesc, notificationParameter))
               }
             }
@@ -94,7 +93,12 @@ object CompactionTask {
     override def run(): Unit = {
       try {
         val table = LakeSoulTable.forPath(path)
-        table.compaction(partitionDesc)
+        val partitions = partitionDesc.split(",").map(
+          partition => {
+            partition.replace("=", "='") + "'"
+          }
+        ).mkString(" and ")
+        table.compaction(partitions)
       } catch {
         case e: Exception => throw e
       } finally {