[SPARK-51097] [SS] Re-introduce RocksDB state store's last uploaded snapshot version instance metrics

[zecookiez]

What changes were proposed in this pull request?

SPARK-51097

#50161 recently had to revert the changes in #49816 due to instance metrics showing up on SparkUI, causing excessive clutter. This PR aims to re-introduce the new instance metrics while incorporating the fix in #50157.

The main difference between this and the original PR are:

• Distinguishing metrics and instance metrics as a whole, so that SparkPlan does not accidentally pick up hundreds of instance metrics
• Adding a test to the suite to verify that no nodes in the execution plan contain the instance metrics we just introduced. Since SparkPlan nodes store metrics as strings, distinguishing between metric types isn't possible in this context. Thus, this test will only check for the specific metric we just introduced.
• Small edit to another test to verify instance metric behavior as well (see here)

Line-by-line:

• statefulOperators.scala line 206-232 is one of the main changes. Instead of adding instance metrics to all metrics, we lazy initialize the instance metrics alongside but keep the two SQLMetric maps separate.
• statefulOperators.scala line 348 to use the new instance metric map
• statefulOperators.scala line 363-364 to use the new instance metric map
• statefulOperators.scala line 449-457 to set instance metrics in the new map instead (parameter variable renamed due to conflict with the new map name)
• RocksDBStateStoreIntegrationSuite line 523-561 to verify instance metrics don't appear in SparkPlan
• Small edit to a pre-existing test in RocksDBStateStoreIntegrationSuite line 127 to instead verify instance metrics still show up in progress reports

Before the fix (note that metrics are sorted lexicographically):

After including the fix:

Why are the changes needed?

From #49816:
There's currently a lack of observability into state store specific maintenance information, notably metrics of the last snapshot version uploaded. This affects the ability to identify performance degradation issues behind maintenance tasks and more as described in SPARK-51097.

Does this PR introduce any user-facing change?

From #49816:
There will be some new metrics displayed from StreamingQueryProgress:

Streaming query made progress: {
  ...
  "stateOperators" : [ {
    ...
    "customMetrics" : {
      ...
      "SnapshotLastUploaded.partition_0_default" : 2,
      "SnapshotLastUploaded.partition_12_default" : 10,
      "SnapshotLastUploaded.partition_8_default" : 10,
      ...
    }
  } ],
  "sources" : ...,
  "sink" : ...
}

How was this patch tested?

Five tests are added to RocksDBStateStoreIntegrationSuite:

• The first four are identical to the tests from [SPARK-51097] [SS] Adding state store instance metrics for last uploaded snapshot version in RocksDB #49816, which verify metrics are properly updating using custom faulty stores and different types of stateful queries (deduplicate and join).
• The fifth test goes through the generated execution plan for stateful queries to make sure no instance metric shows up.

I additionally manually verified these metrics on SparkUI (refer to screenshots above with and without the fix)

Was this patch authored or co-authored using generative AI tooling?

No.