liviazhu opened a new pull request, #56793: URL: https://github.com/apache/spark/pull/56793
### What changes were proposed in this pull request? This implements decoupled state store maintenance: it splits the single maintenance operation into independent snapshot and cleanup operations and reworks the maintenance scheduler so the two cannot starve each other and so that closing a provider cannot race in-flight maintenance. The change is organized as four logical steps (one commit each): 1. **Split maintenance into snapshot and cleanup operations.** `StateStoreProvider` gains `doSnapshotMaintenance()` and `doCleanupMaintenance()` alongside `doMaintenance()` (which now calls both), implemented for the HDFS and RocksDB providers. A `MaintenanceOpRequest` tag is added to the `unloadedProvidersToClose` queue. 2. **Decouple the maintenance scheduler.** Snapshot and cleanup are submitted as separate tasks per provider, tracked by two independent partition sets (`snapshotPartitions` / `cleanupPartitions`). The query thread queues a provider for close instead of submitting maintenance directly; the scheduler routes queued work via the op-request tag. 3. **Add an RW lock, dual pools, and a scheduler trigger.** Each provider gets a fair `ReentrantReadWriteLock`: maintenance holds the read lock and close holds the write lock, so close waits for in-flight maintenance and a maintenance op skips itself when a close is in progress. The single maintenance pool is split into separate snapshot (high-priority) and cleanup (low-priority) pools, sized by a new `snapshotToCleanupThreadRatio` config, so one operation type cannot starve the other. `MaintenanceTask.triggerNow` lets the query thread trigger an immediate scheduler cycle when it queues a provider, instead of waiting for the next periodic tick. 4. **Remove the now-unused `FromTaskThread` maintenance source** and replace it with an explicit `nextOp` on the submitted task to control post-completion behavior (enqueue the remaining op, or close the provider). ### Why are the changes needed? The previous maintenance model ran snapshotting and cleanup as a single operation on a shared thread pool, so a slow or blocked operation of one kind could starve the other, and close could race with in-flight maintenance on the same provider. Decoupling the two operations, bounding concurrency per provider, and guarding close with a read/write lock addresses these issues. ### Does this PR introduce _any_ user-facing change? Yes, two internal configurations change: - New `spark.sql.streaming.stateStore.snapshotToCleanupThreadRatio` (default `0.5`): the fraction of maintenance threads allocated to the snapshot pool; the remainder go to the cleanup pool. - `spark.sql.streaming.stateStore.numStateStoreMaintenanceThreads` is now the total split across both pools, with a minimum of 2 (previously 1); its default changes from `max(cores / 4, 1)` to `max(cores / 4, 2)`. Both are internal configuration knobs; there is no change to query results. ### How was this patch tested? - New `StateStoreDecoupledMaintenanceSuite` (HDFS and RocksDB variants) covering the split operations, the decoupled scheduler, the RW lock, the dual pools, `triggerNow`, and the full queue-to-close lifecycle. - Updated `StateStoreSuite`, `StateStoreInstanceMetricSuite`, and `RocksDBSuite`. - `sql/core` compiles and the suites above pass locally. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code (Opus 4.8) This pull request and its description were written by Isaac. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
