adding purge methods

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/AsyncLogPurge.scala

@@ -42,13 +42,16 @@ trait AsyncLogPurge extends Logging {
 
   protected def purge(threshold: Long): Unit
 
+  protected def purgeOldest(): Unit
+
   protected lazy val useAsyncPurge: Boolean = sparkSession.conf.get(SQLConf.ASYNC_LOG_PURGE)
 
   protected def purgeAsync(batchId: Long): Unit = {
     if (purgeRunning.compareAndSet(false, true)) {
       asyncPurgeExecutorService.execute(() => {
         try {
           purge(batchId - minLogEntriesToMaintain)
+          purgeOldest()
         } catch {
           case throwable: Throwable =>
             logError("Encountered error while performing async log purge", throwable)

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/HDFSMetadataLog.scala

@@ -335,6 +335,19 @@ class HDFSMetadataLog[T <: AnyRef : ClassTag](
     }
   }
 
+  def purgeOldest(minEntriesToMaintain: Int): Unit = {
+    val batchIds = listBatches.sorted
+    if (batchIds.length > minEntriesToMaintain) {
+      val filesToDelete = batchIds.take(batchIds.length - minEntriesToMaintain)
+      filesToDelete.foreach { batchId =>
+        val path = batchIdToPath(batchId)
+        fileManager.delete(path)
+        if (metadataCacheEnabled) batchCache.remove(batchId)
+        logTrace(s"Removed metadata log file: $path")
+      }
+    }
+  }
+
   /** List the available batches on file system. */
   protected def listBatches: Array[Long] = {
     val batchIds = fileManager.list(metadataPath, batchFilesFilter)

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/MicroBatchExecution.scala

@@ -663,6 +663,7 @@ class MicroBatchExecution(
             purgeAsync(execCtx.batchId)
           } else {
             purge(execCtx.batchId - minLogEntriesToMaintain)
+            purgeOldest()
           }
         }
       }

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamExecution.scala

@@ -694,7 +694,11 @@ abstract class StreamExecution(
     logDebug(s"Purging metadata at threshold=$threshold")
     offsetLog.purge(threshold)
     commitLog.purge(threshold)
-    operatorStateMetadataLogs.foreach(_._2.purge(threshold))
+  }
+
+  protected def purgeOldest(): Unit = {
+    operatorStateMetadataLogs.foreach(
+      _._2.purgeOldest(minLogEntriesToMaintain))
   }
 }