You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by do...@apache.org on 2026/04/17 21:56:17 UTC
(spark) branch master updated: [SPARK-56523][SS] Fix RocksDB to close Statistics native handle
This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 0cdbb43624d0 [SPARK-56523][SS] Fix RocksDB to close Statistics native handle
0cdbb43624d0 is described below
commit 0cdbb43624d07fc6ed10586bd20fcb52f39f1685
Author: Pablo Langa <so...@gmail.com>
AuthorDate: Fri Apr 17 14:56:06 2026 -0700
[SPARK-56523][SS] Fix RocksDB to close Statistics native handle
### What changes were proposed in this pull request?
Whilst running Spark tests in our environments, we detected unexpected memory usage; upon analysing the issue in depth, we found that the problem lies in the fact that the statistics object created here is not being released from memory
https://github.com/apache/spark/blob/5d491f62748b4b9c34bc3b5bd7390f7b5ca75053/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala#L151.
Only the options are closed at the close
https://github.com/apache/spark/blob/5d491f62748b4b9c34bc3b5bd7390f7b5ca75053/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala#L2046
which does not release the statistics.
To simulate the problem, I have created a small Java application in which a statistics object is created and destroyed in each iteration, just as is currently done in Spark (https://github.com/planga82/rocksdb-statistics-leak)
```
options.setStatistics(new Statistics());
…
options.close()
```
And another one using the fix
```
Statistics nativeStats = new Statistics();
Options options = new Options().setStatistics(nativeStats);
…
nativeStats.close();
options.close();
```
The test result is as follows
```
RSS Growth (80 iterations, 8-core container)
┌───────┬───────────┬────────────┬─────────────────────────────────────┐
│ Mode │ Start RSS │ End RSS │ Growth │
├───────┼───────────┼────────────┼─────────────────────────────────────┤
│ LEAK │ 62,484 kB │ 110,180 kB │ +44.8 MB (+560 kB/iter, linear) │
├───────┼───────────┼────────────┼─────────────────────────────────────┤
│ FIXED │ 64,244 kB │ 70,344 kB │ +6.0 MB │
└───────┴───────────┴────────────┴─────────────────────────────────────┘
```
A heap dump was generated in both cases using jemalloc, and statistics objects are only visible in the Leak test
```
rocksdb::port::cacheline_aligned_alloc(unsigned long)
└─ rocksdb::CoreLocalArray<StatisticsImpl::StatisticsData>::CoreLocalArray()
└─ rocksdb::StatisticsImpl::StatisticsImpl(shared_ptr<Statistics>)
└─ rocksdb::StatisticsJni::StatisticsJni(...)
└─ Java_org_rocksdb_Statistics_newStatistics___3BJ
```
### Why are the changes needed?
To fix a memory leak
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
Running Spark tests in my environment and monitoring memory usage
### Was this patch authored or co-authored using generative AI tooling?
No
Closes #55391 from planga82/fix/rocksdb_memory_leak.
Authored-by: Pablo Langa <so...@gmail.com>
Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
.../org/apache/spark/sql/execution/streaming/state/RocksDB.scala | 5 +++--
1 file changed, 3 insertions(+), 2 deletions(-)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala
index 67dd7527537f..5b4f1d3084ea 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala
@@ -148,8 +148,8 @@ class RocksDB(
}
private val dbLogger = createLogger() // for forwarding RocksDB native logs to log4j
- rocksDbOptions.setStatistics(new Statistics())
- private val nativeStats = rocksDbOptions.statistics()
+ private val nativeStats = new Statistics()
+ rocksDbOptions.setStatistics(nativeStats)
private val workingDir = createTempDir("workingDir")
@@ -2043,6 +2043,7 @@ class RocksDB(
readOptions.close()
writeOptions.close()
flushOptions.close()
+ nativeStats.close()
rocksDbOptions.close()
dbLogger.close()
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org