[SPARK-44809][SS] Remove unused RocksDB custom metrics for pause/writ…

…eBatch ### What changes were proposed in this pull request? Remove unused RocksDB custom metrics for pause/writeBatch ### Why are the changes needed? Remove unused metrics from query progress ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? Modified existing tests ``` [info] Run completed in 23 seconds, 724 milliseconds. [info] Total number of tests run: 11 [info] Suites: completed 1, aborted 0 [info] Tests: succeeded 11, failed 0, canceled 0, ignored 0, pending 0 [info] All tests passed. [success] Total time: 53 s, completed Aug 14, 2023, 3:30:56 PM ``` Closes apache#42491 from anishshri-db/task/SPARK-44809. Authored-by: Anish Shrigondekar <anish.shrigondekar@databricks.com> Signed-off-by: Jungtaek Lim <kabhwan.opensource@gmail.com>
ueshin · Aug 16, 2023 · 2be20e5 · 2be20e5
1 parent ac567b3
commit 2be20e5
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 10 deletions.
diff --git a/...main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala b/...main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala
@@ -126,10 +126,8 @@ private[sql] class RocksDBStateStoreProvider
         CUSTOM_METRIC_PUT_TIME -> sumNativeOpsLatencyMillis("put"),
         CUSTOM_METRIC_GET_COUNT -> nativeOpsCount("get"),
         CUSTOM_METRIC_PUT_COUNT -> nativeOpsCount("put"),
-        CUSTOM_METRIC_WRITEBATCH_TIME -> commitLatencyMs("writeBatch"),
         CUSTOM_METRIC_FLUSH_TIME -> commitLatencyMs("flush"),
         CUSTOM_METRIC_COMMIT_COMPACT_TIME -> commitLatencyMs("compact"),
-        CUSTOM_METRIC_PAUSE_TIME -> commitLatencyMs("pauseBg"),
         CUSTOM_METRIC_CHECKPOINT_TIME -> commitLatencyMs("checkpoint"),
         CUSTOM_METRIC_FILESYNC_TIME -> commitLatencyMs("fileSync"),
         CUSTOM_METRIC_BYTES_COPIED -> rocksDBMetrics.bytesCopied,
@@ -270,14 +268,10 @@ object RocksDBStateStoreProvider {
     "rocksdbPutCount", "RocksDB: number of put calls")
 
   // Commit latency detailed breakdown
-  val CUSTOM_METRIC_WRITEBATCH_TIME = StateStoreCustomTimingMetric(
-    "rocksdbCommitWriteBatchLatency", "RocksDB: commit - write batch time")
   val CUSTOM_METRIC_FLUSH_TIME = StateStoreCustomTimingMetric(
     "rocksdbCommitFlushLatency", "RocksDB: commit - flush time")
   val CUSTOM_METRIC_COMMIT_COMPACT_TIME = StateStoreCustomTimingMetric(
     "rocksdbCommitCompactLatency", "RocksDB: commit - compact time")
-  val CUSTOM_METRIC_PAUSE_TIME = StateStoreCustomTimingMetric(
-    "rocksdbCommitPauseLatency", "RocksDB: commit - pause bg time")
   val CUSTOM_METRIC_CHECKPOINT_TIME = StateStoreCustomTimingMetric(
     "rocksdbCommitCheckpointLatency", "RocksDB: commit - checkpoint time")
   val CUSTOM_METRIC_FILESYNC_TIME = StateStoreCustomTimingMetric(
@@ -332,8 +326,8 @@ object RocksDBStateStoreProvider {
 
   val ALL_CUSTOM_METRICS = Seq(
     CUSTOM_METRIC_SST_FILE_SIZE, CUSTOM_METRIC_GET_TIME, CUSTOM_METRIC_PUT_TIME,
-    CUSTOM_METRIC_WRITEBATCH_TIME, CUSTOM_METRIC_FLUSH_TIME, CUSTOM_METRIC_COMMIT_COMPACT_TIME,
-    CUSTOM_METRIC_PAUSE_TIME, CUSTOM_METRIC_CHECKPOINT_TIME, CUSTOM_METRIC_FILESYNC_TIME,
+    CUSTOM_METRIC_FLUSH_TIME, CUSTOM_METRIC_COMMIT_COMPACT_TIME,
+    CUSTOM_METRIC_CHECKPOINT_TIME, CUSTOM_METRIC_FILESYNC_TIME,
     CUSTOM_METRIC_BYTES_COPIED, CUSTOM_METRIC_FILES_COPIED, CUSTOM_METRIC_FILES_REUSED,
     CUSTOM_METRIC_ZIP_FILE_BYTES_UNCOMPRESSED, CUSTOM_METRIC_GET_COUNT, CUSTOM_METRIC_PUT_COUNT,
     CUSTOM_METRIC_BLOCK_CACHE_MISS, CUSTOM_METRIC_BLOCK_CACHE_HITS, CUSTOM_METRIC_BYTES_READ,

diff --git a/...la/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreIntegrationSuite.scala b/...la/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreIntegrationSuite.scala
@@ -94,8 +94,8 @@ class RocksDBStateStoreIntegrationSuite extends StreamTest
             val stateOperatorMetrics = nextProgress.stateOperators(0)
             assert(JavaConverters.asScalaSet(stateOperatorMetrics.customMetrics.keySet) === Set(
               "rocksdbGetLatency", "rocksdbCommitCompactLatency", "rocksdbBytesCopied",
-              "rocksdbPutLatency", "rocksdbCommitPauseLatency", "rocksdbFilesReused",
-              "rocksdbCommitWriteBatchLatency", "rocksdbFilesCopied", "rocksdbSstFileSize",
+              "rocksdbPutLatency", "rocksdbFilesReused",
+              "rocksdbFilesCopied", "rocksdbSstFileSize",
               "rocksdbCommitCheckpointLatency", "rocksdbZipFileBytesUncompressed",
               "rocksdbCommitFlushLatency", "rocksdbCommitFileSyncLatencyMs", "rocksdbGetCount",
               "rocksdbPutCount", "rocksdbTotalBytesRead", "rocksdbTotalBytesWritten",