[Spark] Optimize Clean Task sql (#551)

* optimize clean sql Signed-off-by: fphantam <[email protected]> * deal with case when count is 0 Signed-off-by: fphantam <[email protected]> * optimize sql Signed-off-by: fphantam <[email protected]> --------- Signed-off-by: fphantam <[email protected]>
lakesoul-io · Oct 25, 2024 · 199df38 · 199df38
1 parent d233ac0
commit 199df38
Show file tree

Hide file tree

Showing 2 changed files with 27 additions and 32 deletions.
diff --git a/lakesoul-spark/src/main/scala/com/dmetasoul/lakesoul/spark/clean/CleanExpiredData.scala b/lakesoul-spark/src/main/scala/com/dmetasoul/lakesoul/spark/clean/CleanExpiredData.scala
@@ -68,7 +68,6 @@ object CleanExpiredData {
         tableOnlySaveOnceCompaction = onlySaveOnceCompaction
       }
 
-      val latestCompactionTimestampBeforeRedundantTTL = getLatestCompactionTimestamp(tableId, partitionDesc, tableRedundantTTL, spark)
       println("========== deal with new partition ========== ")
       println("tableId:" + tableId)
       println("partitionDesc:" + partitionDesc)
@@ -100,6 +99,7 @@ object CleanExpiredData {
           }
         }
       } else {
+        val latestCompactionTimestampBeforeRedundantTTL = getLatestCompactionTimestamp(tableId, partitionDesc, tableRedundantTTL, spark)
         println("last compactionTimestamp before expiration:" + latestCompactionTimestampBeforeRedundantTTL)
         //no compaction action
         if (latestCompactionTimestampBeforeRedundantTTL == 0L) {
@@ -214,19 +214,12 @@ object CleanExpiredData {
     val sql =
       s"""
          |DELETE FROM partition_info
-         |WHERE (table_id,partition_desc,version)
-         |IN
-         |(SELECT
-         |    table_id,
-         |    partition_desc,
-         |    version
-         |FROM partition_info
          |WHERE
-         |    timestamp < $deadTimestamp
-         |AND
          |    table_id = '$tableId'
          |AND
-         |    partition_desc='$partitionDesc')
+         |    partition_desc='$partitionDesc'
+         |AND
+         |    timestamp < $deadTimestamp
          |""".stripMargin
     val stmt = conn.prepareStatement(sql)
     stmt.execute()
@@ -236,13 +229,14 @@ object CleanExpiredData {
     val sql =
       s"""
          |SELECT
-         |    max(timestamp) max_time
+         |    timestamp max_time
          |from
          |    partition_info
          |where
          |    table_id='$table_id'
          |AND partition_desc = '$partitionDesc'
-         |
+         |ORDER BY
+         |    version DESC
          |""".stripMargin
     sqlToDataframe(sql, spark).select("max_time").first().getLong(0)
   }
@@ -253,7 +247,7 @@ object CleanExpiredData {
       val expiredDateZeroTimeMils = getExpiredDateZeroTimeStamp(expiredDaysField.toString.toInt)
       val sql =
         s"""
-           |SELECT DISTINCT ON (table_id)
+           |SELECT
            |        table_id,
            |        commit_op,
            |        timestamp
@@ -266,7 +260,7 @@ object CleanExpiredData {
            |        AND timestamp < '$expiredDateZeroTimeMils'
            |    ORDER BY
            |        table_id,
-           |        timestamp DESC
+           |        version DESC
            |""".stripMargin
       if (sqlToDataframe(sql, spark).count() > 0)
         latestTimestampMils = sqlToDataframe(sql, spark).select("timestamp").first().getLong(0)
@@ -278,7 +272,7 @@ object CleanExpiredData {
     val redundantDataTTlZeroTimeMils = getExpiredDateZeroTimeStamp(redundantDataTTL.toString.toInt)
     val sql =
       s"""
-         |SELECT DISTINCT ON (table_id)
+         |SELECT
          |        table_id,
          |        commit_op,
          |        timestamp
@@ -289,9 +283,6 @@ object CleanExpiredData {
          |        AND partition_desc = '$partitionDesc'
          |        AND table_id = '$table_id'
          |        AND timestamp > '$redundantDataTTlZeroTimeMils'
-         |    ORDER BY
-         |        table_id,
-         |        timestamp DESC
          |""".stripMargin
     if (sqlToDataframe(sql, spark).count() > 0) {
       true

diff --git a/lakesoul-spark/src/main/scala/com/dmetasoul/lakesoul/spark/clean/CleanOldCompaction.scala b/lakesoul-spark/src/main/scala/com/dmetasoul/lakesoul/spark/clean/CleanOldCompaction.scala
@@ -41,10 +41,11 @@ object CleanOldCompaction {
     val sql =
       s"""
          |WITH expiredCommit AS (
-         |    SELECT DISTINCT ON (table_id)
+         |    SELECT
          |        table_id,
          |        commit_op,
          |        partition_desc,
+         |        snapshot,
          |        timestamp
          |    FROM partition_info
          |    WHERE commit_op = 'CompactionCommit'
@@ -58,14 +59,15 @@ object CleanOldCompaction {
          |FROM (
          |    SELECT file_ops
          |    FROM data_commit_info
-         |    WHERE commit_id IN (
-         |        SELECT DISTINCT unnest(pi.snapshot) AS commit_id
-         |        FROM partition_info pi
-         |        LEFT JOIN expiredCommit ec ON pi.table_id = ec.table_id
-         |        WHERE pi.timestamp < ec.timestamp
-         |            AND pi.commit_op = 'CompactionCommit'
-         |            AND pi.partition_desc = ec.partition_desc
-         |    )
+         |    WHERE table_id= '$tableId'
+         |        AND partition_desc = '$partitionDesc'
+         |        AND commit_id IN (
+         |            SELECT DISTINCT unnest(snapshot)
+         |            FROM (
+         |              SELECT snapshot FROM expiredCommit
+         |              LIMIT (SELECT CASE WHEN COUNT(1) = 0 THEN 1 ELSE COUNT(1) END - 1 FROM expiredCommit) OFFSET 1
+         |              ) t_limit
+         |        )
          |) t
          |CROSS JOIN LATERAL (
          |    SELECT
@@ -81,10 +83,12 @@ object CleanOldCompaction {
 
     })
     pathSet.foreach(p => {
-      val path = new Path(p)
-      val sessionHadoopConf = spark.sessionState.newHadoopConf()
-      val fs = path.getFileSystem(sessionHadoopConf)
-      fs.delete(path, true)
+      if (p != null && p != "") {
+        val path = new Path(p)
+        val sessionHadoopConf = spark.sessionState.newHadoopConf()
+        val fs = path.getFileSystem(sessionHadoopConf)
+        fs.delete(path, true)
+      }
     })
   }