You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "gnanda (via GitHub)" <gi...@apache.org> on 2026/02/28 01:00:54 UTC
[PR] [SPARK-55728][SS] Introduce RocksDB conf for file checksum threadpool size [spark]
gnanda opened a new pull request, #54529:
URL: https://github.com/apache/spark/pull/54529
### What changes were proposed in this pull request?
Introduce a conf for the currently hardcoded file checksum thread pool size
### Why are the changes needed?
Add a conf
### Does this PR introduce _any_ user-facing change?
no
### How was this patch tested?
Unit tests
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code 2.1.58
--
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
Re: [PR] [SPARK-55728][SS] Introduce RocksDB conf for file checksum threadpool size [spark]
Posted by "gnanda (via GitHub)" <gi...@apache.org>.
gnanda commented on code in PR #54529:
URL: https://github.com/apache/spark/pull/54529#discussion_r2873856329
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -3684,6 +3684,16 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val STATE_STORE_ROCKSDB_FILE_CHECKSUM_THREAD_POOL_SIZE =
+ buildConf("spark.sql.streaming.stateStore.rocksdb.fileChecksumThreadPoolSize")
+ .internal()
+ .doc("Number of threads used to compute file checksums concurrently when uploading " +
+ "RocksDB state store checkpoints (e.g. main file and checksum file).")
Review Comment:
Done
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/checkpointing/ChecksumCheckpointFileManager.scala:
##########
@@ -150,12 +152,13 @@ class ChecksumCheckpointFileManager(
val numThreads: Int,
val skipCreationIfFileMissingChecksum: Boolean)
extends CheckpointFileManager with Logging {
- assert(numThreads % 2 == 0, "numThreads must be a multiple of 2, we need 1 for the main file" +
+ assert(numThreads == 1 || numThreads % 2 == 0,
Review Comment:
Done
--
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
Re: [PR] [SPARK-55728][SS] Introduce RocksDB conf for file checksum threadpool size [spark]
Posted by "micheal-o (via GitHub)" <gi...@apache.org>.
micheal-o commented on code in PR #54529:
URL: https://github.com/apache/spark/pull/54529#discussion_r2875610836
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -3695,6 +3684,18 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val STATE_STORE_FILE_CHECKSUM_THREAD_POOL_SIZE =
+ buildConf("spark.sql.streaming.stateStore.fileChecksumThreadPoolSize")
+ .internal()
+ .doc("Number of threads used to compute file checksums concurrently when uploading " +
+ "state store checkpoints (e.g. main file and checksum file). " +
Review Comment:
nit: Number of threads used to read/write files and their corresponding checksum files concurrently
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/checkpointing/ChecksumCheckpointFileManager.scala:
##########
@@ -150,14 +153,33 @@ class ChecksumCheckpointFileManager(
val numThreads: Int,
val skipCreationIfFileMissingChecksum: Boolean)
extends CheckpointFileManager with Logging {
- assert(numThreads % 2 == 0, "numThreads must be a multiple of 2, we need 1 for the main file" +
- "and another for the checksum file")
+ assert(numThreads >= 0, "numThreads must be a non-negative integer")
import ChecksumCheckpointFileManager._
- // This allows us to concurrently read/write the main file and checksum file
- private val threadPool = ExecutionContext.fromExecutorService(
- ThreadUtils.newDaemonFixedThreadPool(numThreads, s"${this.getClass.getSimpleName}-Thread"))
+ // Thread pool for concurrent execution, or None when numThreads == 0 (sequential/inline mode).
+ private val threadPool: Option[ExecutionContextExecutorService] =
Review Comment:
nit: rename to `threadPoolOpt` to make it obvious
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/checkpointing/ChecksumCheckpointFileManager.scala:
##########
@@ -150,14 +153,33 @@ class ChecksumCheckpointFileManager(
val numThreads: Int,
val skipCreationIfFileMissingChecksum: Boolean)
extends CheckpointFileManager with Logging {
- assert(numThreads % 2 == 0, "numThreads must be a multiple of 2, we need 1 for the main file" +
- "and another for the checksum file")
+ assert(numThreads >= 0, "numThreads must be a non-negative integer")
import ChecksumCheckpointFileManager._
- // This allows us to concurrently read/write the main file and checksum file
- private val threadPool = ExecutionContext.fromExecutorService(
- ThreadUtils.newDaemonFixedThreadPool(numThreads, s"${this.getClass.getSimpleName}-Thread"))
+ // Thread pool for concurrent execution, or None when numThreads == 0 (sequential/inline mode).
Review Comment:
please lets also retain the original comment here
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/checkpointing/ChecksumCheckpointFileManager.scala:
##########
@@ -150,14 +153,33 @@ class ChecksumCheckpointFileManager(
val numThreads: Int,
val skipCreationIfFileMissingChecksum: Boolean)
extends CheckpointFileManager with Logging {
- assert(numThreads % 2 == 0, "numThreads must be a multiple of 2, we need 1 for the main file" +
- "and another for the checksum file")
+ assert(numThreads >= 0, "numThreads must be a non-negative integer")
import ChecksumCheckpointFileManager._
- // This allows us to concurrently read/write the main file and checksum file
- private val threadPool = ExecutionContext.fromExecutorService(
- ThreadUtils.newDaemonFixedThreadPool(numThreads, s"${this.getClass.getSimpleName}-Thread"))
+ // Thread pool for concurrent execution, or None when numThreads == 0 (sequential/inline mode).
+ private val threadPool: Option[ExecutionContextExecutorService] =
+ if (numThreads == 0) None
+ else Some(ExecutionContext.fromExecutorService(
+ ThreadUtils.newDaemonFixedThreadPool(numThreads, s"${this.getClass.getSimpleName}-Thread")))
+
+ // ExecutionContext to pass to stream classes: uses the thread pool or a caller-runs context.
+ private val streamContext: ExecutionContext = threadPool.getOrElse(new ExecutionContext {
+ override def execute(runnable: Runnable): Unit = runnable.run()
Review Comment:
nit: comment: `This will execute the runnable synchronously`
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/checkpointing/ChecksumCheckpointFileManager.scala:
##########
@@ -150,14 +153,33 @@ class ChecksumCheckpointFileManager(
val numThreads: Int,
val skipCreationIfFileMissingChecksum: Boolean)
extends CheckpointFileManager with Logging {
- assert(numThreads % 2 == 0, "numThreads must be a multiple of 2, we need 1 for the main file" +
- "and another for the checksum file")
+ assert(numThreads >= 0, "numThreads must be a non-negative integer")
import ChecksumCheckpointFileManager._
- // This allows us to concurrently read/write the main file and checksum file
- private val threadPool = ExecutionContext.fromExecutorService(
- ThreadUtils.newDaemonFixedThreadPool(numThreads, s"${this.getClass.getSimpleName}-Thread"))
+ // Thread pool for concurrent execution, or None when numThreads == 0 (sequential/inline mode).
+ private val threadPool: Option[ExecutionContextExecutorService] =
+ if (numThreads == 0) None
+ else Some(ExecutionContext.fromExecutorService(
+ ThreadUtils.newDaemonFixedThreadPool(numThreads, s"${this.getClass.getSimpleName}-Thread")))
+
+ // ExecutionContext to pass to stream classes: uses the thread pool or a caller-runs context.
Review Comment:
nit: you mean input/outStream? Lets make it clear
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/checkpointing/ChecksumCheckpointFileManager.scala:
##########
@@ -127,12 +128,14 @@ case class ChecksumFile(path: Path) {
* orphan checksum files. If using this, it is your responsibility
* to clean up the potential orphan checksum files.
* @param numThreads This is the number of threads to use for the thread pool, for reading/writing
- * files. To avoid blocking, if the file manager instance is being used by a
- * single thread, then you can set this to 2 (one thread for main file, another
- * for checksum file).
- * If file manager is shared by multiple threads, you can set it to
- * number of threads using file manager * 2.
- * Setting this differently can lead to file operation being blocked waiting for
+ * files. Must be a non-negative integer.
+ * Setting this to 0 disables the thread pool and runs all operations
+ * sequentially on the calling thread (no concurrency).
+ * To avoid blocking with a single concurrent caller, set this to 2 (one thread
Review Comment:
> To avoid blocking
For this remaining statements, lets use the original statement. This new one is not very clear.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/checkpointing/ChecksumCheckpointFileManager.scala:
##########
@@ -150,14 +153,33 @@ class ChecksumCheckpointFileManager(
val numThreads: Int,
val skipCreationIfFileMissingChecksum: Boolean)
extends CheckpointFileManager with Logging {
- assert(numThreads % 2 == 0, "numThreads must be a multiple of 2, we need 1 for the main file" +
- "and another for the checksum file")
+ assert(numThreads >= 0, "numThreads must be a non-negative integer")
import ChecksumCheckpointFileManager._
- // This allows us to concurrently read/write the main file and checksum file
- private val threadPool = ExecutionContext.fromExecutorService(
- ThreadUtils.newDaemonFixedThreadPool(numThreads, s"${this.getClass.getSimpleName}-Thread"))
+ // Thread pool for concurrent execution, or None when numThreads == 0 (sequential/inline mode).
+ private val threadPool: Option[ExecutionContextExecutorService] =
+ if (numThreads == 0) None
+ else Some(ExecutionContext.fromExecutorService(
+ ThreadUtils.newDaemonFixedThreadPool(numThreads, s"${this.getClass.getSimpleName}-Thread")))
+
+ // ExecutionContext to pass to stream classes: uses the thread pool or a caller-runs context.
+ private val streamContext: ExecutionContext = threadPool.getOrElse(new ExecutionContext {
Review Comment:
why is this named `streamContext`?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/checkpointing/ChecksumCheckpointFileManager.scala:
##########
@@ -150,14 +153,33 @@ class ChecksumCheckpointFileManager(
val numThreads: Int,
val skipCreationIfFileMissingChecksum: Boolean)
extends CheckpointFileManager with Logging {
- assert(numThreads % 2 == 0, "numThreads must be a multiple of 2, we need 1 for the main file" +
- "and another for the checksum file")
+ assert(numThreads >= 0, "numThreads must be a non-negative integer")
import ChecksumCheckpointFileManager._
- // This allows us to concurrently read/write the main file and checksum file
- private val threadPool = ExecutionContext.fromExecutorService(
- ThreadUtils.newDaemonFixedThreadPool(numThreads, s"${this.getClass.getSimpleName}-Thread"))
+ // Thread pool for concurrent execution, or None when numThreads == 0 (sequential/inline mode).
+ private val threadPool: Option[ExecutionContextExecutorService] =
+ if (numThreads == 0) None
+ else Some(ExecutionContext.fromExecutorService(
+ ThreadUtils.newDaemonFixedThreadPool(numThreads, s"${this.getClass.getSimpleName}-Thread")))
+
+ // ExecutionContext to pass to stream classes: uses the thread pool or a caller-runs context.
+ private val streamContext: ExecutionContext = threadPool.getOrElse(new ExecutionContext {
+ override def execute(runnable: Runnable): Unit = runnable.run()
+ override def reportFailure(cause: Throwable): Unit = throw cause
+ })
+
+ /**
+ * Schedules a computation on the thread pool, or runs it directly on the calling thread
+ * if numThreads == 0 (sequential mode).
+ */
+ private def scheduleOrRun[T](f: => T): Future[T] = threadPool match {
+ case None =>
+ try Future.successful(f)
Review Comment:
nit: `try {}`
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/checkpointing/ChecksumCheckpointFileManager.scala:
##########
@@ -150,14 +153,33 @@ class ChecksumCheckpointFileManager(
val numThreads: Int,
val skipCreationIfFileMissingChecksum: Boolean)
extends CheckpointFileManager with Logging {
- assert(numThreads % 2 == 0, "numThreads must be a multiple of 2, we need 1 for the main file" +
- "and another for the checksum file")
+ assert(numThreads >= 0, "numThreads must be a non-negative integer")
import ChecksumCheckpointFileManager._
- // This allows us to concurrently read/write the main file and checksum file
- private val threadPool = ExecutionContext.fromExecutorService(
- ThreadUtils.newDaemonFixedThreadPool(numThreads, s"${this.getClass.getSimpleName}-Thread"))
+ // Thread pool for concurrent execution, or None when numThreads == 0 (sequential/inline mode).
+ private val threadPool: Option[ExecutionContextExecutorService] =
+ if (numThreads == 0) None
+ else Some(ExecutionContext.fromExecutorService(
+ ThreadUtils.newDaemonFixedThreadPool(numThreads, s"${this.getClass.getSimpleName}-Thread")))
+
+ // ExecutionContext to pass to stream classes: uses the thread pool or a caller-runs context.
+ private val streamContext: ExecutionContext = threadPool.getOrElse(new ExecutionContext {
+ override def execute(runnable: Runnable): Unit = runnable.run()
+ override def reportFailure(cause: Throwable): Unit = throw cause
+ })
+
+ /**
+ * Schedules a computation on the thread pool, or runs it directly on the calling thread
+ * if numThreads == 0 (sequential mode).
+ */
+ private def scheduleOrRun[T](f: => T): Future[T] = threadPool match {
Review Comment:
Why have `scheduleOrRun` and the synchronous Exec Context above? Why not only use the sync Exec context. Not sure I get why we need both
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/StateStoreSuite.scala:
##########
@@ -1547,6 +1547,35 @@ class StateStoreSuite extends StateStoreSuiteBase[HDFSBackedStateStoreProvider]
}
}
+ test("fileChecksumThreadPoolSize propagates to ChecksumCheckpointFileManager") {
+ Seq(0, 1, 6).foreach { numThreads =>
+ withTempDir { dir =>
+ val sqlConf = SQLConf.get.clone()
+ sqlConf.setConfString(SQLConf.STREAMING_CHECKPOINT_FILE_CHECKSUM_ENABLED.key, "true")
+ sqlConf.setConfString(
+ SQLConf.STATE_STORE_FILE_CHECKSUM_THREAD_POOL_SIZE.key, numThreads.toString)
+
+ val provider = newStoreProvider(
+ opId = Random.nextInt(), partition = 0, dir = dir.getCanonicalPath,
+ sqlConfOpt = Some(sqlConf))
+ val fileManagerMethod = PrivateMethod[CheckpointFileManager](Symbol("fm"))
+ val fm = provider invokePrivate fileManagerMethod()
+
+ assert(fm.isInstanceOf[ChecksumCheckpointFileManager])
+ assert(fm.asInstanceOf[ChecksumCheckpointFileManager].numThreads === numThreads)
+ provider.close()
+ }
+ }
+ }
Review Comment:
Lets also add test case with threadpool disabled to make sure using the state store works i.e. load, write, get, commit. Reload.. etc. You can also do concurrent store.commit and maintenance, to make sure they both finish
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/ChecksumCheckpointFileManagerSuite.scala:
##########
@@ -250,6 +250,30 @@ abstract class ChecksumCheckpointFileManagerSuite extends CheckpointFileManagerT
checksumFmWithoutFallback.close()
}
}
+
+ test("numThreads = 0 disables thread pool") {
+ withTempHadoopPath { basePath =>
+ val fm = new ChecksumCheckpointFileManager(
+ createNoChecksumManager(basePath),
+ allowConcurrentDelete = true,
+ numThreads = 0,
+ skipCreationIfFileMissingChecksum = false)
+ fm.close()
Review Comment:
This isn't testing much. We should write and read a file to confirm that main and checksum files are written and read
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/HDFSBackedStateStoreProvider.scala:
##########
@@ -533,11 +533,10 @@ private[sql] class HDFSBackedStateStoreProvider extends StateStoreProvider with
mgr,
// Allowing this for perf, since we do orphan checksum file cleanup in maintenance anyway
allowConcurrentDelete = true,
- // We need 2 threads per fm caller to avoid blocking
- // (one for main file and another for checksum file).
- // Since this fm is used by both query task and maintenance thread,
- // then we need 2 * 2 = 4 threads.
- numThreads = 4,
+ // To avoid blocking, we need 2 threads per fm caller (one for main file, one for checksum
+ // file). Since this fm is used by both query task and maintenance thread, the recommended
+ // default is 2 * 2 = 4 threads. A value of 0 disables the thread pool (sequential mode).
+ numThreads = storeConf.fileChecksumThreadPoolSize,
Review Comment:
lets also do the logging for this too right
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/StateStoreSuite.scala:
##########
@@ -1547,6 +1547,35 @@ class StateStoreSuite extends StateStoreSuiteBase[HDFSBackedStateStoreProvider]
}
}
+ test("fileChecksumThreadPoolSize propagates to ChecksumCheckpointFileManager") {
Review Comment:
The test cases are currently under StateStoreSuite, if you look at the class def, it is for HDFS store only. This should be under `StateStoreSuiteBase` so that is runs for rocksdb too
--
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
Re: [PR] [SPARK-55728][SS] Introduce RocksDB conf for file checksum threadpool size [spark]
Posted by "micheal-o (via GitHub)" <gi...@apache.org>.
micheal-o commented on code in PR #54529:
URL: https://github.com/apache/spark/pull/54529#discussion_r2875632815
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/checkpointing/ChecksumCheckpointFileManager.scala:
##########
@@ -279,18 +301,20 @@ class ChecksumCheckpointFileManager(
}
override def close(): Unit = {
- threadPool.shutdown()
- // Wait a bit for it to finish up in case there is any ongoing work
- // Can consider making this timeout configurable, if needed
- val timeoutMs = 500
- if (!threadPool.awaitTermination(timeoutMs, TimeUnit.MILLISECONDS)) {
- logWarning(log"Thread pool did not shutdown after ${MDC(TIMEOUT, timeoutMs)} ms," +
- log" forcing shutdown")
- threadPool.shutdownNow() // stop the executing tasks
-
- // Wait a bit for the threads to respond
- if (!threadPool.awaitTermination(timeoutMs, TimeUnit.MILLISECONDS)) {
- logError(log"Thread pool did not terminate")
+ threadPool.foreach { pool =>
+ pool.shutdown()
Review Comment:
@holdenk the `threadPool` here is an `Option`, hence at most one value
--
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
Re: [PR] [SPARK-55728][SS] Introduce conf for file checksum threadpool size and support disabling the threadpool [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on code in PR #54529:
URL: https://github.com/apache/spark/pull/54529#discussion_r2943009102
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/HDFSBackedStateStoreProvider.scala:
##########
@@ -529,15 +529,19 @@ private[sql] class HDFSBackedStateStoreProvider extends StateStoreProvider with
private[state] lazy val fm = {
val mgr = CheckpointFileManager.create(baseDir, hadoopConf)
if (storeConf.checkpointFileChecksumEnabled) {
+ // To avoid blocking, we need 2 threads per fm caller (one for main file, one for checksum
+ // file). Since this fm is used by both query task and maintenance thread, the recommended
+ // default is 2 * 2 = 4 threads. A value of 0 disables the thread pool (sequential mode).
+ val numThreads = storeConf.fileChecksumThreadPoolSize
+ if (numThreads < 4) {
+ logWarning(s"fileChecksumThreadPoolSize is set to $numThreads, which is below the " +
Review Comment:
Ah ok - yea there could be different file checksums right. Just want to make it obvious that we are referring to state-store-file-checksum. So maybe something like this
`Thread pool size for name=state-store-file-checksum is set to $numThreads....`
--
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
Re: [PR] [SPARK-55728][SS] Introduce RocksDB conf for file checksum threadpool size and support disabling the threadpool [spark]
Posted by "gnanda (via GitHub)" <gi...@apache.org>.
gnanda commented on code in PR #54529:
URL: https://github.com/apache/spark/pull/54529#discussion_r2881392205
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/checkpointing/ChecksumCheckpointFileManager.scala:
##########
@@ -150,14 +153,33 @@ class ChecksumCheckpointFileManager(
val numThreads: Int,
val skipCreationIfFileMissingChecksum: Boolean)
extends CheckpointFileManager with Logging {
- assert(numThreads % 2 == 0, "numThreads must be a multiple of 2, we need 1 for the main file" +
- "and another for the checksum file")
+ assert(numThreads >= 0, "numThreads must be a non-negative integer")
import ChecksumCheckpointFileManager._
- // This allows us to concurrently read/write the main file and checksum file
- private val threadPool = ExecutionContext.fromExecutorService(
- ThreadUtils.newDaemonFixedThreadPool(numThreads, s"${this.getClass.getSimpleName}-Thread"))
+ // Thread pool for concurrent execution, or None when numThreads == 0 (sequential/inline mode).
+ private val threadPool: Option[ExecutionContextExecutorService] =
+ if (numThreads == 0) None
+ else Some(ExecutionContext.fromExecutorService(
+ ThreadUtils.newDaemonFixedThreadPool(numThreads, s"${this.getClass.getSimpleName}-Thread")))
+
+ // ExecutionContext to pass to stream classes: uses the thread pool or a caller-runs context.
+ private val streamContext: ExecutionContext = threadPool.getOrElse(new ExecutionContext {
+ override def execute(runnable: Runnable): Unit = runnable.run()
+ override def reportFailure(cause: Throwable): Unit = throw cause
+ })
+
+ /**
+ * Schedules a computation on the thread pool, or runs it directly on the calling thread
+ * if numThreads == 0 (sequential mode).
+ */
+ private def scheduleOrRun[T](f: => T): Future[T] = threadPool match {
Review Comment:
Removed scheduleOrRun.
Not too familar with scala, and AI suggests weird things...
--
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
Re: [PR] [SPARK-55728][SS] Introduce RocksDB conf for file checksum threadpool size and support disabling the threadpool [spark]
Posted by "gnanda (via GitHub)" <gi...@apache.org>.
gnanda commented on code in PR #54529:
URL: https://github.com/apache/spark/pull/54529#discussion_r2881392739
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/checkpointing/ChecksumCheckpointFileManager.scala:
##########
@@ -150,14 +153,33 @@ class ChecksumCheckpointFileManager(
val numThreads: Int,
val skipCreationIfFileMissingChecksum: Boolean)
extends CheckpointFileManager with Logging {
- assert(numThreads % 2 == 0, "numThreads must be a multiple of 2, we need 1 for the main file" +
- "and another for the checksum file")
+ assert(numThreads >= 0, "numThreads must be a non-negative integer")
import ChecksumCheckpointFileManager._
- // This allows us to concurrently read/write the main file and checksum file
- private val threadPool = ExecutionContext.fromExecutorService(
- ThreadUtils.newDaemonFixedThreadPool(numThreads, s"${this.getClass.getSimpleName}-Thread"))
+ // Thread pool for concurrent execution, or None when numThreads == 0 (sequential/inline mode).
+ private val threadPool: Option[ExecutionContextExecutorService] =
+ if (numThreads == 0) None
+ else Some(ExecutionContext.fromExecutorService(
+ ThreadUtils.newDaemonFixedThreadPool(numThreads, s"${this.getClass.getSimpleName}-Thread")))
+
+ // ExecutionContext to pass to stream classes: uses the thread pool or a caller-runs context.
+ private val streamContext: ExecutionContext = threadPool.getOrElse(new ExecutionContext {
+ override def execute(runnable: Runnable): Unit = runnable.run()
+ override def reportFailure(cause: Throwable): Unit = throw cause
+ })
+
+ /**
+ * Schedules a computation on the thread pool, or runs it directly on the calling thread
+ * if numThreads == 0 (sequential mode).
+ */
+ private def scheduleOrRun[T](f: => T): Future[T] = threadPool match {
+ case None =>
+ try Future.successful(f)
Review Comment:
No longer applicable
--
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
Re: [PR] [SPARK-55728][SS] Introduce RocksDB conf for file checksum threadpool size [spark]
Posted by "gnanda (via GitHub)" <gi...@apache.org>.
gnanda commented on code in PR #54529:
URL: https://github.com/apache/spark/pull/54529#discussion_r2873852823
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -3684,6 +3684,16 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val STATE_STORE_ROCKSDB_FILE_CHECKSUM_THREAD_POOL_SIZE =
+ buildConf("spark.sql.streaming.stateStore.rocksdb.fileChecksumThreadPoolSize")
+ .internal()
+ .doc("Number of threads used to compute file checksums concurrently when uploading " +
+ "RocksDB state store checkpoints (e.g. main file and checksum file).")
+ .version("4.1.0")
Review Comment:
Done
--
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
Re: [PR] [SPARK-55728][SS] Introduce RocksDB conf for file checksum threadpool size and support disabling the threadpool [spark]
Posted by "gnanda (via GitHub)" <gi...@apache.org>.
gnanda commented on code in PR #54529:
URL: https://github.com/apache/spark/pull/54529#discussion_r2881385434
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -3695,6 +3684,18 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val STATE_STORE_FILE_CHECKSUM_THREAD_POOL_SIZE =
+ buildConf("spark.sql.streaming.stateStore.fileChecksumThreadPoolSize")
+ .internal()
+ .doc("Number of threads used to compute file checksums concurrently when uploading " +
+ "state store checkpoints (e.g. main file and checksum file). " +
Review Comment:
Done
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/checkpointing/ChecksumCheckpointFileManager.scala:
##########
@@ -127,12 +128,14 @@ case class ChecksumFile(path: Path) {
* orphan checksum files. If using this, it is your responsibility
* to clean up the potential orphan checksum files.
* @param numThreads This is the number of threads to use for the thread pool, for reading/writing
- * files. To avoid blocking, if the file manager instance is being used by a
- * single thread, then you can set this to 2 (one thread for main file, another
- * for checksum file).
- * If file manager is shared by multiple threads, you can set it to
- * number of threads using file manager * 2.
- * Setting this differently can lead to file operation being blocked waiting for
+ * files. Must be a non-negative integer.
+ * Setting this to 0 disables the thread pool and runs all operations
+ * sequentially on the calling thread (no concurrency).
+ * To avoid blocking with a single concurrent caller, set this to 2 (one thread
Review Comment:
Reverted
--
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
Re: [PR] [SPARK-55728][SS] Introduce RocksDB conf for file checksum threadpool size and support disabling the threadpool [spark]
Posted by "gnanda (via GitHub)" <gi...@apache.org>.
gnanda commented on code in PR #54529:
URL: https://github.com/apache/spark/pull/54529#discussion_r2881388366
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/checkpointing/ChecksumCheckpointFileManager.scala:
##########
@@ -150,14 +153,33 @@ class ChecksumCheckpointFileManager(
val numThreads: Int,
val skipCreationIfFileMissingChecksum: Boolean)
extends CheckpointFileManager with Logging {
- assert(numThreads % 2 == 0, "numThreads must be a multiple of 2, we need 1 for the main file" +
- "and another for the checksum file")
+ assert(numThreads >= 0, "numThreads must be a non-negative integer")
import ChecksumCheckpointFileManager._
- // This allows us to concurrently read/write the main file and checksum file
- private val threadPool = ExecutionContext.fromExecutorService(
- ThreadUtils.newDaemonFixedThreadPool(numThreads, s"${this.getClass.getSimpleName}-Thread"))
+ // Thread pool for concurrent execution, or None when numThreads == 0 (sequential/inline mode).
+ private val threadPool: Option[ExecutionContextExecutorService] =
+ if (numThreads == 0) None
+ else Some(ExecutionContext.fromExecutorService(
+ ThreadUtils.newDaemonFixedThreadPool(numThreads, s"${this.getClass.getSimpleName}-Thread")))
+
+ // ExecutionContext to pass to stream classes: uses the thread pool or a caller-runs context.
Review Comment:
Done
--
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
Re: [PR] [SPARK-55728][SS] Introduce RocksDB conf for file checksum threadpool size and support disabling the threadpool [spark]
Posted by "gnanda (via GitHub)" <gi...@apache.org>.
gnanda commented on code in PR #54529:
URL: https://github.com/apache/spark/pull/54529#discussion_r2881384827
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/ChecksumCheckpointFileManagerSuite.scala:
##########
@@ -250,6 +250,30 @@ abstract class ChecksumCheckpointFileManagerSuite extends CheckpointFileManagerT
checksumFmWithoutFallback.close()
}
}
+
+ test("numThreads = 0 disables thread pool") {
+ withTempHadoopPath { basePath =>
+ val fm = new ChecksumCheckpointFileManager(
+ createNoChecksumManager(basePath),
+ allowConcurrentDelete = true,
+ numThreads = 0,
+ skipCreationIfFileMissingChecksum = false)
+ fm.close()
Review Comment:
Added the checks
--
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
Re: [PR] [SPARK-55728][SS] Introduce conf for file checksum threadpool size and support disabling the threadpool [spark]
Posted by "micheal-o (via GitHub)" <gi...@apache.org>.
micheal-o commented on code in PR #54529:
URL: https://github.com/apache/spark/pull/54529#discussion_r2942993616
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/HDFSBackedStateStoreProvider.scala:
##########
@@ -529,15 +529,19 @@ private[sql] class HDFSBackedStateStoreProvider extends StateStoreProvider with
private[state] lazy val fm = {
val mgr = CheckpointFileManager.create(baseDir, hadoopConf)
if (storeConf.checkpointFileChecksumEnabled) {
+ // To avoid blocking, we need 2 threads per fm caller (one for main file, one for checksum
+ // file). Since this fm is used by both query task and maintenance thread, the recommended
+ // default is 2 * 2 = 4 threads. A value of 0 disables the thread pool (sequential mode).
+ val numThreads = storeConf.fileChecksumThreadPoolSize
+ if (numThreads < 4) {
+ logWarning(s"fileChecksumThreadPoolSize is set to $numThreads, which is below the " +
Review Comment:
@anishshri-db the log already mentioned file checksum threadpool and the log line would include the state store provider id. So which extra info do you need 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
Re: [PR] [SPARK-55728][SS] Introduce RocksDB conf for file checksum threadpool size and support disabling the threadpool [spark]
Posted by "gnanda (via GitHub)" <gi...@apache.org>.
gnanda commented on code in PR #54529:
URL: https://github.com/apache/spark/pull/54529#discussion_r2881397325
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/StateStoreSuite.scala:
##########
@@ -1547,6 +1547,35 @@ class StateStoreSuite extends StateStoreSuiteBase[HDFSBackedStateStoreProvider]
}
}
+ test("fileChecksumThreadPoolSize propagates to ChecksumCheckpointFileManager") {
Review Comment:
Moved
--
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
Re: [PR] [SPARK-55728][SS] Introduce RocksDB conf for file checksum threadpool size and support disabling the threadpool [spark]
Posted by "gnanda (via GitHub)" <gi...@apache.org>.
gnanda commented on code in PR #54529:
URL: https://github.com/apache/spark/pull/54529#discussion_r2881398146
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/StateStoreSuite.scala:
##########
@@ -1547,6 +1547,35 @@ class StateStoreSuite extends StateStoreSuiteBase[HDFSBackedStateStoreProvider]
}
}
+ test("fileChecksumThreadPoolSize propagates to ChecksumCheckpointFileManager") {
+ Seq(0, 1, 6).foreach { numThreads =>
+ withTempDir { dir =>
+ val sqlConf = SQLConf.get.clone()
+ sqlConf.setConfString(SQLConf.STREAMING_CHECKPOINT_FILE_CHECKSUM_ENABLED.key, "true")
+ sqlConf.setConfString(
+ SQLConf.STATE_STORE_FILE_CHECKSUM_THREAD_POOL_SIZE.key, numThreads.toString)
+
+ val provider = newStoreProvider(
+ opId = Random.nextInt(), partition = 0, dir = dir.getCanonicalPath,
+ sqlConfOpt = Some(sqlConf))
+ val fileManagerMethod = PrivateMethod[CheckpointFileManager](Symbol("fm"))
+ val fm = provider invokePrivate fileManagerMethod()
+
+ assert(fm.isInstanceOf[ChecksumCheckpointFileManager])
+ assert(fm.asInstanceOf[ChecksumCheckpointFileManager].numThreads === numThreads)
+ provider.close()
+ }
+ }
+ }
Review Comment:
Added
--
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
Re: [PR] [SPARK-55728][SS] Introduce RocksDB conf for file checksum threadpool size [spark]
Posted by "holdenk (via GitHub)" <gi...@apache.org>.
holdenk commented on code in PR #54529:
URL: https://github.com/apache/spark/pull/54529#discussion_r2875354449
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/checkpointing/ChecksumCheckpointFileManager.scala:
##########
@@ -127,12 +128,14 @@ case class ChecksumFile(path: Path) {
* orphan checksum files. If using this, it is your responsibility
* to clean up the potential orphan checksum files.
* @param numThreads This is the number of threads to use for the thread pool, for reading/writing
- * files. To avoid blocking, if the file manager instance is being used by a
- * single thread, then you can set this to 2 (one thread for main file, another
- * for checksum file).
- * If file manager is shared by multiple threads, you can set it to
- * number of threads using file manager * 2.
- * Setting this differently can lead to file operation being blocked waiting for
+ * files. Must be a non-negative integer.
+ * Setting this to 0 disables the thread pool and runs all operations
+ * sequentially on the calling thread (no concurrency).
+ * To avoid blocking with a single concurrent caller, set this to 2 (one thread
+ * for main file, another for checksum file).
+ * If file manager is shared by multiple threads, set it to
+ * number of concurrent callers * 2.
Review Comment:
Should this go over in the doc config? Also how would a user figure out how many threads are going to be accessing it?
--
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
Re: [PR] [SPARK-55728][SS] Introduce RocksDB conf for file checksum threadpool size [spark]
Posted by "holdenk (via GitHub)" <gi...@apache.org>.
holdenk commented on code in PR #54529:
URL: https://github.com/apache/spark/pull/54529#discussion_r2875357640
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/checkpointing/ChecksumCheckpointFileManager.scala:
##########
@@ -279,18 +301,20 @@ class ChecksumCheckpointFileManager(
}
override def close(): Unit = {
- threadPool.shutdown()
- // Wait a bit for it to finish up in case there is any ongoing work
- // Can consider making this timeout configurable, if needed
- val timeoutMs = 500
- if (!threadPool.awaitTermination(timeoutMs, TimeUnit.MILLISECONDS)) {
- logWarning(log"Thread pool did not shutdown after ${MDC(TIMEOUT, timeoutMs)} ms," +
- log" forcing shutdown")
- threadPool.shutdownNow() // stop the executing tasks
-
- // Wait a bit for the threads to respond
- if (!threadPool.awaitTermination(timeoutMs, TimeUnit.MILLISECONDS)) {
- logError(log"Thread pool did not terminate")
+ threadPool.foreach { pool =>
+ pool.shutdown()
Review Comment:
Maybe silly but would it make sense to do this first foreach and then the subsequent blocking checkin? It's only during shutdown so not a big deal.
--
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
Re: [PR] [SPARK-55728][SS] Introduce RocksDB conf for file checksum threadpool size [spark]
Posted by "gnanda (via GitHub)" <gi...@apache.org>.
gnanda commented on code in PR #54529:
URL: https://github.com/apache/spark/pull/54529#discussion_r2873862825
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBSuite.scala:
##########
@@ -2995,6 +2995,63 @@ class RocksDBSuite extends AlsoTestWithRocksDBFeatures with SharedSparkSession
}
}
+ test("RocksDB: fileChecksumThreadPoolSize propagates to ChecksumCheckpointFileManager") {
+ withTempDir { dir =>
+ val sqlConf = SQLConf.get.clone()
+ sqlConf.setConfString(
+ SQLConf.STREAMING_CHECKPOINT_FILE_CHECKSUM_ENABLED.key, "true")
+ sqlConf.setConfString(
+ SQLConf.STATE_STORE_ROCKSDB_FILE_CHECKSUM_THREAD_POOL_SIZE.key, "6")
+
+ val dbConf = RocksDBConf(StateStoreConf(sqlConf))
+ assert(dbConf.fileChecksumThreadPoolSize === 6)
+
+ val remoteDir = dir.getCanonicalPath
+ withDB(remoteDir, conf = dbConf) { db =>
+ // Access the fm (lazy val) on fileManager via PrivateMethodTester
+ val fileManagerMethod = PrivateMethod[CheckpointFileManager](Symbol("fm"))
+ val fm = db.fileManager invokePrivate fileManagerMethod()
+
+ // Verify it's a ChecksumCheckpointFileManager with the right thread count
+ assert(fm.isInstanceOf[ChecksumCheckpointFileManager])
+ assert(fm.asInstanceOf[ChecksumCheckpointFileManager].numThreads === 6)
+ }
+ }
+ }
+
+ test("RocksDB: fileChecksumThreadPoolSize of 1 is valid and propagates correctly") {
+ withTempDir { dir =>
+ val sqlConf = SQLConf.get.clone()
+ sqlConf.setConfString(
+ SQLConf.STREAMING_CHECKPOINT_FILE_CHECKSUM_ENABLED.key, "true")
+ sqlConf.setConfString(
+ SQLConf.STATE_STORE_ROCKSDB_FILE_CHECKSUM_THREAD_POOL_SIZE.key, "1")
Review Comment:
done
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala:
##########
@@ -152,11 +152,10 @@ class RocksDB(
private val workingDir = createTempDir("workingDir")
- // We need 2 threads per fm caller to avoid blocking
- // (one for main file and another for checksum file).
- // Since this fm is used by both query task and maintenance thread,
- // then we need 2 * 2 = 4 threads.
- protected val fileChecksumThreadPoolSize: Option[Int] = Some(4)
+ // To avoid blocking, we need 2 threads per fm caller (one for main file, one for checksum file).
+ // Since this fm is used by both query task and maintenance thread, the recommended default is
+ // 2 * 2 = 4 threads. A value of 1 disables concurrency (sequential execution).
+ protected val fileChecksumThreadPoolSize: Option[Int] = Some(conf.fileChecksumThreadPoolSize)
Review Comment:
done
--
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
Re: [PR] [SPARK-55728][SS] Introduce RocksDB conf for file checksum threadpool size and support disabling the threadpool [spark]
Posted by "gnanda (via GitHub)" <gi...@apache.org>.
gnanda commented on code in PR #54529:
URL: https://github.com/apache/spark/pull/54529#discussion_r2921065601
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/StateStoreSuite.scala:
##########
@@ -2465,6 +2465,107 @@ abstract class StateStoreSuiteBase[ProviderClass <: StateStoreProvider]
}
}
+ test("fileChecksumThreadPoolSize propagates to ChecksumCheckpointFileManager") {
+ Seq(0, 1, 6).foreach { numThreads =>
+ val storeId = StateStoreId(newDir(), 0L, 0)
+ withSQLConf(
+ SQLConf.STREAMING_CHECKPOINT_FILE_CHECKSUM_ENABLED.key -> "true",
+ SQLConf.STATE_STORE_FILE_CHECKSUM_THREAD_POOL_SIZE.key -> numThreads.toString) {
+ tryWithProviderResource(newStoreProviderWithClonedConf(storeId)) { provider =>
+ val fmMethod = PrivateMethod[CheckpointFileManager](Symbol("fm"))
+ val fm = provider match {
+ case hdfs: HDFSBackedStateStoreProvider =>
+ hdfs.fm
+ case rocksdb: RocksDBStateStoreProvider =>
+ rocksdb.rocksDB.fileManager invokePrivate fmMethod()
+ case _ =>
+ fail(s"Unexpected provider type: ${provider.getClass.getName}")
+ }
+ assert(fm.isInstanceOf[ChecksumCheckpointFileManager])
+ assert(fm.asInstanceOf[ChecksumCheckpointFileManager].numThreads === numThreads)
+ }
+ }
+ }
+ }
+
+ test("STATE_STORE_FILE_CHECKSUM_THREAD_POOL_SIZE: invalid negative value is rejected") {
+ val sqlConf = SQLConf.get.clone()
+ val ex = intercept[IllegalArgumentException] {
+ sqlConf.setConfString(SQLConf.STATE_STORE_FILE_CHECKSUM_THREAD_POOL_SIZE.key, "-1")
+ }
+ assert(ex.getMessage.contains("Must be a non-negative integer"))
+ }
+
+ test("fileChecksumThreadPoolSize = 0 supports sequential I/O (load, write, commit, reload)") {
+ val storeId = StateStoreId(newDir(), 0L, 0)
+ withSQLConf(
+ SQLConf.STREAMING_CHECKPOINT_FILE_CHECKSUM_ENABLED.key -> "true",
+ SQLConf.STATE_STORE_FILE_CHECKSUM_THREAD_POOL_SIZE.key -> "0") {
+ // Write some state with sequential mode enabled
+ tryWithProviderResource(newStoreProviderWithClonedConf(storeId)) { provider =>
+ val store = provider.getStore(0)
+ put(store, "a", 0, 1)
+ put(store, "b", 0, 2)
+ store.commit()
+ provider.doMaintenance()
Review Comment:
Removed here, and changed the concurrent test to enable `changelogCheckpointing` (and reduced minDeltasForSnapshot so there's a snapshot) so that maintenance will upload the snapshot
--
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
Re: [PR] [SPARK-55728][SS] Introduce RocksDB conf for file checksum threadpool size [spark]
Posted by "micheal-o (via GitHub)" <gi...@apache.org>.
micheal-o commented on code in PR #54529:
URL: https://github.com/apache/spark/pull/54529#discussion_r2875611153
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/checkpointing/ChecksumCheckpointFileManager.scala:
##########
@@ -127,12 +128,14 @@ case class ChecksumFile(path: Path) {
* orphan checksum files. If using this, it is your responsibility
* to clean up the potential orphan checksum files.
* @param numThreads This is the number of threads to use for the thread pool, for reading/writing
- * files. To avoid blocking, if the file manager instance is being used by a
- * single thread, then you can set this to 2 (one thread for main file, another
- * for checksum file).
- * If file manager is shared by multiple threads, you can set it to
- * number of threads using file manager * 2.
- * Setting this differently can lead to file operation being blocked waiting for
+ * files. Must be a non-negative integer.
+ * Setting this to 0 disables the thread pool and runs all operations
+ * sequentially on the calling thread (no concurrency).
+ * To avoid blocking with a single concurrent caller, set this to 2 (one thread
Review Comment:
> To avoid blocking
For this remaining statements, lets use the original statement. This new one is not very clear.
--
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
Re: [PR] [SPARK-55728][SS] Introduce RocksDB conf for file checksum threadpool size [spark]
Posted by "gnanda (via GitHub)" <gi...@apache.org>.
gnanda commented on code in PR #54529:
URL: https://github.com/apache/spark/pull/54529#discussion_r2875128839
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala:
##########
@@ -152,11 +152,17 @@ class RocksDB(
private val workingDir = createTempDir("workingDir")
- // We need 2 threads per fm caller to avoid blocking
- // (one for main file and another for checksum file).
- // Since this fm is used by both query task and maintenance thread,
- // then we need 2 * 2 = 4 threads.
- protected val fileChecksumThreadPoolSize: Option[Int] = Some(4)
+ // To avoid blocking, we need 2 threads per fm caller (one for main file, one for checksum file).
+ // Since this fm is used by both query task and maintenance thread, the recommended default is
+ // 2 * 2 = 4 threads. A value of 0 disables the thread pool (sequential execution).
+ protected val fileChecksumThreadPoolSize: Option[Int] = {
+ val size = conf.fileChecksumThreadPoolSize
+ if (size != 4) {
Review Comment:
Ah good catch; missed that from the AI change. Done
--
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
Re: [PR] [SPARK-55728][SS] Introduce conf for file checksum threadpool size and support disabling the threadpool [spark]
Posted by "micheal-o (via GitHub)" <gi...@apache.org>.
micheal-o commented on code in PR #54529:
URL: https://github.com/apache/spark/pull/54529#discussion_r2943002084
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -3718,6 +3684,18 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val STATE_STORE_FILE_CHECKSUM_THREAD_POOL_SIZE =
+ buildConf("spark.sql.streaming.stateStore.fileChecksumThreadPoolSize")
+ .internal()
+ .doc("Number of threads used to read/write files and their corresponding checksum files " +
+ "concurrently. Set to 0 to disable the thread pool and run operations sequentially on " +
Review Comment:
@anishshri-db just incase someone is running on a very constrained cluster and wouldn't mind the latency increase. They are prioritizing cost.
--
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
Re: [PR] [SPARK-55728][SS] Introduce conf for file checksum threadpool size and support disabling the threadpool [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on code in PR #54529:
URL: https://github.com/apache/spark/pull/54529#discussion_r2942974869
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/HDFSBackedStateStoreProvider.scala:
##########
@@ -529,15 +529,19 @@ private[sql] class HDFSBackedStateStoreProvider extends StateStoreProvider with
private[state] lazy val fm = {
val mgr = CheckpointFileManager.create(baseDir, hadoopConf)
if (storeConf.checkpointFileChecksumEnabled) {
+ // To avoid blocking, we need 2 threads per fm caller (one for main file, one for checksum
+ // file). Since this fm is used by both query task and maintenance thread, the recommended
+ // default is 2 * 2 = 4 threads. A value of 0 disables the thread pool (sequential mode).
+ val numThreads = storeConf.fileChecksumThreadPoolSize
+ if (numThreads < 4) {
Review Comment:
Lets define the constant in one place and then reuse in all places
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala:
##########
@@ -152,11 +152,17 @@ class RocksDB(
private val workingDir = createTempDir("workingDir")
- // We need 2 threads per fm caller to avoid blocking
- // (one for main file and another for checksum file).
- // Since this fm is used by both query task and maintenance thread,
- // then we need 2 * 2 = 4 threads.
- protected val fileChecksumThreadPoolSize: Option[Int] = Some(4)
+ // To avoid blocking, we need 2 threads per fm caller (one for main file, one for checksum file).
+ // Since this fm is used by both query task and maintenance thread, the recommended default is
+ // 2 * 2 = 4 threads. A value of 0 disables the thread pool (sequential execution).
+ protected val fileChecksumThreadPoolSize: Option[Int] = {
+ val size = conf.fileChecksumThreadPoolSize
+ if (size < 4) {
Review Comment:
Same here
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/HDFSBackedStateStoreProvider.scala:
##########
@@ -529,15 +529,19 @@ private[sql] class HDFSBackedStateStoreProvider extends StateStoreProvider with
private[state] lazy val fm = {
val mgr = CheckpointFileManager.create(baseDir, hadoopConf)
if (storeConf.checkpointFileChecksumEnabled) {
+ // To avoid blocking, we need 2 threads per fm caller (one for main file, one for checksum
+ // file). Since this fm is used by both query task and maintenance thread, the recommended
+ // default is 2 * 2 = 4 threads. A value of 0 disables the thread pool (sequential mode).
+ val numThreads = storeConf.fileChecksumThreadPoolSize
+ if (numThreads < 4) {
+ logWarning(s"fileChecksumThreadPoolSize is set to $numThreads, which is below the " +
Review Comment:
nit: can we provide more info around which thread pool we refer to here, to make debugging based on executor logs easier
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/checkpointing/ChecksumCheckpointFileManager.scala:
##########
@@ -150,14 +152,26 @@ class ChecksumCheckpointFileManager(
val numThreads: Int,
val skipCreationIfFileMissingChecksum: Boolean)
extends CheckpointFileManager with Logging {
- assert(numThreads % 2 == 0, "numThreads must be a multiple of 2, we need 1 for the main file" +
- "and another for the checksum file")
+ assert(numThreads >= 0, "numThreads must be a non-negative integer")
import ChecksumCheckpointFileManager._
// This allows us to concurrently read/write the main file and checksum file
- private val threadPool = ExecutionContext.fromExecutorService(
- ThreadUtils.newDaemonFixedThreadPool(numThreads, s"${this.getClass.getSimpleName}-Thread"))
+ private val threadPoolOpt: Option[ExecutionContextExecutorService] =
+ if (numThreads == 0) None
Review Comment:
nit - can we just add it in braces
```
if (numThreads == 0) {
None
} else {
Some(ExecutionContext.fromExecutorService(
ThreadUtils.newDaemonFixedThreadPool(numThreads, s"${this.getClass.getSimpleName}-Thread")))
}
```
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -3718,6 +3684,18 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val STATE_STORE_FILE_CHECKSUM_THREAD_POOL_SIZE =
+ buildConf("spark.sql.streaming.stateStore.fileChecksumThreadPoolSize")
+ .internal()
+ .doc("Number of threads used to read/write files and their corresponding checksum files " +
+ "concurrently. Set to 0 to disable the thread pool and run operations sequentially on " +
Review Comment:
Why do we want to support disabling the thread pool ?
--
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
Re: [PR] [SPARK-55728][SS] Introduce conf for file checksum threadpool size and support disabling the threadpool [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db closed pull request #54529: [SPARK-55728][SS] Introduce conf for file checksum threadpool size and support disabling the threadpool
URL: https://github.com/apache/spark/pull/54529
--
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
Re: [PR] [SPARK-55728][SS] Introduce RocksDB conf for file checksum threadpool size [spark]
Posted by "gnanda (via GitHub)" <gi...@apache.org>.
gnanda commented on code in PR #54529:
URL: https://github.com/apache/spark/pull/54529#discussion_r2873857911
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateStoreConf.scala:
##########
@@ -116,6 +116,9 @@ class StateStoreConf(
/** Whether file checksum generation and verification is enabled. */
val checkpointFileChecksumEnabled: Boolean = sqlConf.checkpointFileChecksumEnabled
+ /** Number of threads used to compute file checksums concurrently for RocksDB state store. */
Review Comment:
Done
--
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
Re: [PR] [SPARK-55728][SS] Introduce RocksDB conf for file checksum threadpool size and support disabling the threadpool [spark]
Posted by "gnanda (via GitHub)" <gi...@apache.org>.
gnanda commented on code in PR #54529:
URL: https://github.com/apache/spark/pull/54529#discussion_r2921058227
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/HDFSBackedStateStoreProvider.scala:
##########
@@ -533,11 +533,10 @@ private[sql] class HDFSBackedStateStoreProvider extends StateStoreProvider with
mgr,
// Allowing this for perf, since we do orphan checksum file cleanup in maintenance anyway
allowConcurrentDelete = true,
- // We need 2 threads per fm caller to avoid blocking
- // (one for main file and another for checksum file).
- // Since this fm is used by both query task and maintenance thread,
- // then we need 2 * 2 = 4 threads.
- numThreads = 4,
+ // To avoid blocking, we need 2 threads per fm caller (one for main file, one for checksum
+ // file). Since this fm is used by both query task and maintenance thread, the recommended
+ // default is 2 * 2 = 4 threads. A value of 0 disables the thread pool (sequential mode).
+ numThreads = storeConf.fileChecksumThreadPoolSize,
Review Comment:
Done
--
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
Re: [PR] [SPARK-55728][SS] Introduce RocksDB conf for file checksum threadpool size [spark]
Posted by "gnanda (via GitHub)" <gi...@apache.org>.
gnanda commented on code in PR #54529:
URL: https://github.com/apache/spark/pull/54529#discussion_r2873851465
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -3684,6 +3684,16 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val STATE_STORE_ROCKSDB_FILE_CHECKSUM_THREAD_POOL_SIZE =
+ buildConf("spark.sql.streaming.stateStore.rocksdb.fileChecksumThreadPoolSize")
+ .internal()
+ .doc("Number of threads used to compute file checksums concurrently when uploading " +
+ "RocksDB state store checkpoints (e.g. main file and checksum file).")
+ .version("4.1.0")
+ .intConf
+ .checkValue(x => x == 1 || (x > 0 && x % 2 == 0), "Must be 1 or a positive even number")
Review Comment:
Done
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -3684,6 +3684,16 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val STATE_STORE_ROCKSDB_FILE_CHECKSUM_THREAD_POOL_SIZE =
+ buildConf("spark.sql.streaming.stateStore.rocksdb.fileChecksumThreadPoolSize")
+ .internal()
+ .doc("Number of threads used to compute file checksums concurrently when uploading " +
+ "RocksDB state store checkpoints (e.g. main file and checksum file).")
+ .version("4.1.0")
+ .intConf
+ .checkValue(x => x == 1 || (x > 0 && x % 2 == 0), "Must be 1 or a positive even number")
Review Comment:
Done
--
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
Re: [PR] [SPARK-55728][SS] Introduce RocksDB conf for file checksum threadpool size [spark]
Posted by "ericm-db (via GitHub)" <gi...@apache.org>.
ericm-db commented on code in PR #54529:
URL: https://github.com/apache/spark/pull/54529#discussion_r2874923907
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala:
##########
@@ -152,11 +152,17 @@ class RocksDB(
private val workingDir = createTempDir("workingDir")
- // We need 2 threads per fm caller to avoid blocking
- // (one for main file and another for checksum file).
- // Since this fm is used by both query task and maintenance thread,
- // then we need 2 * 2 = 4 threads.
- protected val fileChecksumThreadPoolSize: Option[Int] = Some(4)
+ // To avoid blocking, we need 2 threads per fm caller (one for main file, one for checksum file).
+ // Since this fm is used by both query task and maintenance thread, the recommended default is
+ // 2 * 2 = 4 threads. A value of 0 disables the thread pool (sequential execution).
+ protected val fileChecksumThreadPoolSize: Option[Int] = {
+ val size = conf.fileChecksumThreadPoolSize
+ if (size != 4) {
Review Comment:
Should we not only log this warning if size < 4
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -3684,6 +3684,16 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val STATE_STORE_ROCKSDB_FILE_CHECKSUM_THREAD_POOL_SIZE =
+ buildConf("spark.sql.streaming.stateStore.rocksdb.fileChecksumThreadPoolSize")
+ .internal()
+ .doc("Number of threads used to compute file checksums concurrently when uploading " +
+ "RocksDB state store checkpoints (e.g. main file and checksum file).")
+ .version("4.1.0")
Review Comment:
This should be 4.2.0
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -3684,6 +3684,16 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val STATE_STORE_ROCKSDB_FILE_CHECKSUM_THREAD_POOL_SIZE =
+ buildConf("spark.sql.streaming.stateStore.rocksdb.fileChecksumThreadPoolSize")
+ .internal()
+ .doc("Number of threads used to compute file checksums concurrently when uploading " +
+ "RocksDB state store checkpoints (e.g. main file and checksum file).")
+ .version("4.1.0")
Review Comment:
This should be 4.2.0
--
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
Re: [PR] [SPARK-55728][SS] Introduce RocksDB conf for file checksum threadpool size and support disabling the threadpool [spark]
Posted by "gnanda (via GitHub)" <gi...@apache.org>.
gnanda commented on code in PR #54529:
URL: https://github.com/apache/spark/pull/54529#discussion_r2881389818
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/checkpointing/ChecksumCheckpointFileManager.scala:
##########
@@ -150,14 +153,33 @@ class ChecksumCheckpointFileManager(
val numThreads: Int,
val skipCreationIfFileMissingChecksum: Boolean)
extends CheckpointFileManager with Logging {
- assert(numThreads % 2 == 0, "numThreads must be a multiple of 2, we need 1 for the main file" +
- "and another for the checksum file")
+ assert(numThreads >= 0, "numThreads must be a non-negative integer")
import ChecksumCheckpointFileManager._
- // This allows us to concurrently read/write the main file and checksum file
- private val threadPool = ExecutionContext.fromExecutorService(
- ThreadUtils.newDaemonFixedThreadPool(numThreads, s"${this.getClass.getSimpleName}-Thread"))
+ // Thread pool for concurrent execution, or None when numThreads == 0 (sequential/inline mode).
+ private val threadPool: Option[ExecutionContextExecutorService] =
+ if (numThreads == 0) None
+ else Some(ExecutionContext.fromExecutorService(
+ ThreadUtils.newDaemonFixedThreadPool(numThreads, s"${this.getClass.getSimpleName}-Thread")))
+
+ // ExecutionContext to pass to stream classes: uses the thread pool or a caller-runs context.
+ private val streamContext: ExecutionContext = threadPool.getOrElse(new ExecutionContext {
Review Comment:
changed to `executionContext` (AI chose that original name; wasn't sure if it was a scala thing)
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/checkpointing/ChecksumCheckpointFileManager.scala:
##########
@@ -150,14 +153,33 @@ class ChecksumCheckpointFileManager(
val numThreads: Int,
val skipCreationIfFileMissingChecksum: Boolean)
extends CheckpointFileManager with Logging {
- assert(numThreads % 2 == 0, "numThreads must be a multiple of 2, we need 1 for the main file" +
- "and another for the checksum file")
+ assert(numThreads >= 0, "numThreads must be a non-negative integer")
import ChecksumCheckpointFileManager._
- // This allows us to concurrently read/write the main file and checksum file
- private val threadPool = ExecutionContext.fromExecutorService(
- ThreadUtils.newDaemonFixedThreadPool(numThreads, s"${this.getClass.getSimpleName}-Thread"))
+ // Thread pool for concurrent execution, or None when numThreads == 0 (sequential/inline mode).
+ private val threadPool: Option[ExecutionContextExecutorService] =
+ if (numThreads == 0) None
+ else Some(ExecutionContext.fromExecutorService(
+ ThreadUtils.newDaemonFixedThreadPool(numThreads, s"${this.getClass.getSimpleName}-Thread")))
+
+ // ExecutionContext to pass to stream classes: uses the thread pool or a caller-runs context.
+ private val streamContext: ExecutionContext = threadPool.getOrElse(new ExecutionContext {
+ override def execute(runnable: Runnable): Unit = runnable.run()
Review Comment:
done
--
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
Re: [PR] [SPARK-55728][SS] Introduce RocksDB conf for file checksum threadpool size and support disabling the threadpool [spark]
Posted by "micheal-o (via GitHub)" <gi...@apache.org>.
micheal-o commented on code in PR #54529:
URL: https://github.com/apache/spark/pull/54529#discussion_r2898433220
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/StateStoreSuite.scala:
##########
@@ -2465,6 +2465,107 @@ abstract class StateStoreSuiteBase[ProviderClass <: StateStoreProvider]
}
}
+ test("fileChecksumThreadPoolSize propagates to ChecksumCheckpointFileManager") {
+ Seq(0, 1, 6).foreach { numThreads =>
+ val storeId = StateStoreId(newDir(), 0L, 0)
+ withSQLConf(
+ SQLConf.STREAMING_CHECKPOINT_FILE_CHECKSUM_ENABLED.key -> "true",
+ SQLConf.STATE_STORE_FILE_CHECKSUM_THREAD_POOL_SIZE.key -> numThreads.toString) {
+ tryWithProviderResource(newStoreProviderWithClonedConf(storeId)) { provider =>
+ val fmMethod = PrivateMethod[CheckpointFileManager](Symbol("fm"))
+ val fm = provider match {
+ case hdfs: HDFSBackedStateStoreProvider =>
+ hdfs.fm
+ case rocksdb: RocksDBStateStoreProvider =>
+ rocksdb.rocksDB.fileManager invokePrivate fmMethod()
+ case _ =>
+ fail(s"Unexpected provider type: ${provider.getClass.getName}")
+ }
+ assert(fm.isInstanceOf[ChecksumCheckpointFileManager])
+ assert(fm.asInstanceOf[ChecksumCheckpointFileManager].numThreads === numThreads)
+ }
+ }
+ }
+ }
+
+ test("STATE_STORE_FILE_CHECKSUM_THREAD_POOL_SIZE: invalid negative value is rejected") {
+ val sqlConf = SQLConf.get.clone()
+ val ex = intercept[IllegalArgumentException] {
+ sqlConf.setConfString(SQLConf.STATE_STORE_FILE_CHECKSUM_THREAD_POOL_SIZE.key, "-1")
+ }
+ assert(ex.getMessage.contains("Must be a non-negative integer"))
+ }
+
+ test("fileChecksumThreadPoolSize = 0 supports sequential I/O (load, write, commit, reload)") {
+ val storeId = StateStoreId(newDir(), 0L, 0)
+ withSQLConf(
+ SQLConf.STREAMING_CHECKPOINT_FILE_CHECKSUM_ENABLED.key -> "true",
+ SQLConf.STATE_STORE_FILE_CHECKSUM_THREAD_POOL_SIZE.key -> "0") {
+ // Write some state with sequential mode enabled
+ tryWithProviderResource(newStoreProviderWithClonedConf(storeId)) { provider =>
+ val store = provider.getStore(0)
+ put(store, "a", 0, 1)
+ put(store, "b", 0, 2)
+ store.commit()
+ provider.doMaintenance()
Review Comment:
maintenance isn't uploading anything. We are not producing snapshots to upload. Same for all the other maintenance calls below
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/checkpointing/ChecksumCheckpointFileManager.scala:
##########
@@ -150,14 +152,30 @@ class ChecksumCheckpointFileManager(
val numThreads: Int,
val skipCreationIfFileMissingChecksum: Boolean)
extends CheckpointFileManager with Logging {
- assert(numThreads % 2 == 0, "numThreads must be a multiple of 2, we need 1 for the main file" +
- "and another for the checksum file")
+ assert(numThreads >= 0, "numThreads must be a non-negative integer")
+ if (numThreads < 4) {
+ logWarning(s"numThreads is set to $numThreads, which is below the recommended minimum of 4 " +
Review Comment:
This log shouldn't be here. It's the caller e.g. state store, that knows how many threads it will need
--
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
Re: [PR] [SPARK-55728][SS] Introduce RocksDB conf for file checksum threadpool size and support disabling the threadpool [spark]
Posted by "gnanda (via GitHub)" <gi...@apache.org>.
gnanda commented on code in PR #54529:
URL: https://github.com/apache/spark/pull/54529#discussion_r2921060150
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/checkpointing/ChecksumCheckpointFileManager.scala:
##########
@@ -150,14 +152,30 @@ class ChecksumCheckpointFileManager(
val numThreads: Int,
val skipCreationIfFileMissingChecksum: Boolean)
extends CheckpointFileManager with Logging {
- assert(numThreads % 2 == 0, "numThreads must be a multiple of 2, we need 1 for the main file" +
- "and another for the checksum file")
+ assert(numThreads >= 0, "numThreads must be a non-negative integer")
+ if (numThreads < 4) {
+ logWarning(s"numThreads is set to $numThreads, which is below the recommended minimum of 4 " +
Review Comment:
Sounds good, I added it to both the HDFS and RocksDB implementations instead of 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
Re: [PR] [SPARK-55728][SS] Introduce RocksDB conf for file checksum threadpool size [spark]
Posted by "gnanda (via GitHub)" <gi...@apache.org>.
gnanda commented on code in PR #54529:
URL: https://github.com/apache/spark/pull/54529#discussion_r2873850406
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -3684,6 +3684,16 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val STATE_STORE_ROCKSDB_FILE_CHECKSUM_THREAD_POOL_SIZE =
Review Comment:
Done
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -3684,6 +3684,16 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val STATE_STORE_ROCKSDB_FILE_CHECKSUM_THREAD_POOL_SIZE =
+ buildConf("spark.sql.streaming.stateStore.rocksdb.fileChecksumThreadPoolSize")
Review Comment:
Done
--
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
Re: [PR] [SPARK-55728][SS] Introduce RocksDB conf for file checksum threadpool size [spark]
Posted by "gnanda (via GitHub)" <gi...@apache.org>.
gnanda commented on code in PR #54529:
URL: https://github.com/apache/spark/pull/54529#discussion_r2873861942
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBSuite.scala:
##########
@@ -2995,6 +2995,63 @@ class RocksDBSuite extends AlsoTestWithRocksDBFeatures with SharedSparkSession
}
}
+ test("RocksDB: fileChecksumThreadPoolSize propagates to ChecksumCheckpointFileManager") {
Review Comment:
Done ( I think)
--
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
Re: [PR] [SPARK-55728][SS] Introduce RocksDB conf for file checksum threadpool size [spark]
Posted by "micheal-o (via GitHub)" <gi...@apache.org>.
micheal-o commented on code in PR #54529:
URL: https://github.com/apache/spark/pull/54529#discussion_r2875626446
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/checkpointing/ChecksumCheckpointFileManager.scala:
##########
@@ -127,12 +128,14 @@ case class ChecksumFile(path: Path) {
* orphan checksum files. If using this, it is your responsibility
* to clean up the potential orphan checksum files.
* @param numThreads This is the number of threads to use for the thread pool, for reading/writing
- * files. To avoid blocking, if the file manager instance is being used by a
- * single thread, then you can set this to 2 (one thread for main file, another
- * for checksum file).
- * If file manager is shared by multiple threads, you can set it to
- * number of threads using file manager * 2.
- * Setting this differently can lead to file operation being blocked waiting for
+ * files. Must be a non-negative integer.
+ * Setting this to 0 disables the thread pool and runs all operations
+ * sequentially on the calling thread (no concurrency).
+ * To avoid blocking with a single concurrent caller, set this to 2 (one thread
+ * for main file, another for checksum file).
+ * If file manager is shared by multiple threads, set it to
+ * number of concurrent callers * 2.
Review Comment:
@holdenk No this is an internal detail. The config doc already has the user friendly info. Ideally the user shouldn't be changing this conf. It is just there if they want to disable it or reduce the default threadpool for some reason (e.g. resource constraint).
--
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
Re: [PR] [SPARK-55728][SS] Introduce RocksDB conf for file checksum threadpool size and support disabling the threadpool [spark]
Posted by "gnanda (via GitHub)" <gi...@apache.org>.
gnanda commented on code in PR #54529:
URL: https://github.com/apache/spark/pull/54529#discussion_r2881386451
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/checkpointing/ChecksumCheckpointFileManager.scala:
##########
@@ -150,14 +153,33 @@ class ChecksumCheckpointFileManager(
val numThreads: Int,
val skipCreationIfFileMissingChecksum: Boolean)
extends CheckpointFileManager with Logging {
- assert(numThreads % 2 == 0, "numThreads must be a multiple of 2, we need 1 for the main file" +
- "and another for the checksum file")
+ assert(numThreads >= 0, "numThreads must be a non-negative integer")
import ChecksumCheckpointFileManager._
- // This allows us to concurrently read/write the main file and checksum file
- private val threadPool = ExecutionContext.fromExecutorService(
- ThreadUtils.newDaemonFixedThreadPool(numThreads, s"${this.getClass.getSimpleName}-Thread"))
+ // Thread pool for concurrent execution, or None when numThreads == 0 (sequential/inline mode).
Review Comment:
reverted
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/checkpointing/ChecksumCheckpointFileManager.scala:
##########
@@ -150,14 +153,33 @@ class ChecksumCheckpointFileManager(
val numThreads: Int,
val skipCreationIfFileMissingChecksum: Boolean)
extends CheckpointFileManager with Logging {
- assert(numThreads % 2 == 0, "numThreads must be a multiple of 2, we need 1 for the main file" +
- "and another for the checksum file")
+ assert(numThreads >= 0, "numThreads must be a non-negative integer")
import ChecksumCheckpointFileManager._
- // This allows us to concurrently read/write the main file and checksum file
- private val threadPool = ExecutionContext.fromExecutorService(
- ThreadUtils.newDaemonFixedThreadPool(numThreads, s"${this.getClass.getSimpleName}-Thread"))
+ // Thread pool for concurrent execution, or None when numThreads == 0 (sequential/inline mode).
+ private val threadPool: Option[ExecutionContextExecutorService] =
Review Comment:
Done
--
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