Re: [PR] [SPARK-55493] [SS] Do not mkdirs in streaming checkpoint offset/commit log directory in StateDataSource [spark]

["anishshri-db (via GitHub)" <gi...@apache.org>]

anishshri-db commented on code in PR #54381:
URL: https://github.com/apache/spark/pull/54381#discussion_r2920721131


##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/checkpointing/CommitLog.scala:
##########
@@ -46,8 +46,11 @@ import org.apache.spark.sql.internal.SQLConf
  * line 1: version
  * line 2: metadata (optional json string)
  */
-class CommitLog(sparkSession: SparkSession, path: String)
-  extends HDFSMetadataLog[CommitMetadata](sparkSession, path) {
+class CommitLog(
+    sparkSession: SparkSession,
+    path: String,
+    readOnly: Boolean = false)

Review Comment:
   Is this changing the on-disk format also for commit log ?



##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/checkpointing/HDFSMetadataLog.scala:
##########
@@ -327,6 +331,9 @@ class HDFSMetadataLog[T <: AnyRef : ClassTag](sparkSession: SparkSession, path:
 
   /** List the available batches on file system. */
   protected def listBatches: Array[Long] = {
+    if (!fileManager.exists(metadataPath)) {

Review Comment:
   Why do we need this ?



##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/state/StateDataSourceReadSuite.scala:
##########
@@ -1526,6 +1526,56 @@ class StateDataSourceNoEmptyDirCreationSuite extends StateDataSourceTestBase {
         s"but got: ${e.getClass.getSimpleName}: ${e.getMessage}")
   }
 
+  test("deleted offsets directory is not recreated on read") {
+    withTempDir { tempDir =>
+      val checkpointPath = tempDir.getAbsolutePath
+      runLargeDataStreamingAggregationQuery(checkpointPath)
+
+      val offsetsDir = new File(tempDir, "offsets")
+      assert(offsetsDir.exists(), "Offsets directory should exist after running the query")
+      Utils.deleteRecursively(offsetsDir)
+      assert(!offsetsDir.exists(), "Offsets directory should be deleted")
+
+      val e1 = intercept[Exception] {
+        spark.read
+          .format("statestore")

Review Comment:
   Can we add tests for `state-metadata` source also ?



##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/checkpointing/HDFSMetadataLog.scala:
##########
@@ -351,6 +358,9 @@ class HDFSMetadataLog[T <: AnyRef : ClassTag](sparkSession: SparkSession, path:
    * @return array of batches ids
    */
   def listBatchesOnDisk: Array[Long] = {
+    if (!fileManager.exists(metadataPath)) {
+      return Array.empty

Review Comment:
   same here ?



##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/checkpointing/HDFSMetadataLog.scala:
##########
@@ -48,7 +48,10 @@ import org.apache.spark.util.Utils
  * Note: [[HDFSMetadataLog]] doesn't support S3-like file systems as they don't guarantee listing
  * files in a directory always shows the latest files.
  */
-class HDFSMetadataLog[T <: AnyRef : ClassTag](sparkSession: SparkSession, path: String)
+class HDFSMetadataLog[T <: AnyRef : ClassTag](
+    sparkSession: SparkSession,
+    path: String,
+    readOnly: Boolean = false)

Review Comment:
   same here ?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org