[SPARK-55131][SS] Change the default merge operator delimiter for RocksDB to empty string to concat without delimiter by HeartSaVioR · Pull Request #54083 · apache/spark

HeartSaVioR · 2026-02-01T04:09:12Z

What changes were proposed in this pull request?

This PR proposes to change the default delimiter for the merge operator of RocksDB to an empty string, so that merge operation does not add a delimiter and concat two without any character.

Changing the delimiter isn't compatible with existing checkpoints, so this change is coupled with SQLConf, with known offset log metadata trick, to apply the change only for new streaming queries.

New SQL config: spark.sql.streaming.stateStore.rocksdb.mergeOperatorVersion
Default: 2 ('' as delimiter)
Default for existing checkpoints: 1 (',' as delimiter)

Why are the changes needed?

We found out there is no way to distinguish two cases of 1) put against non-existence value then merge and 2) merge against non-existence value then merge, from the current delimiter. There has been an "implication" that operators do ensure they call merge only when they know the operation is against existing key. This effectively requires GET operation which can be an outstanding performance impact depending on the logic.

Making delimiter to an empty string (none) would eliminate the difference between the two cases, allowing operators to perform blind merge without checking the existence of the key.

Does this PR introduce any user-facing change?

No, the change is internal and there is no user-facing change.

How was this patch tested?

Added UTs.

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

Co-authored by claude-4.5-sonnet

github-actions · 2026-02-01T04:09:21Z

JIRA Issue Information

=== Bug SPARK-55131 ===
Summary: The default delimiter of StringAppendOperator (merge operator for RocksDB) conflicts when merge is used with non-existence value
Assignee: None
Status: Open
Affected: ["4.2.0"]

This comment was automatically generated by GitHub Actions

anishshri-db · 2026-02-05T02:09:45Z

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

+    buildConf("spark.sql.streaming.stateStore.rocksdb.mergeOperatorVersion")
+      .internal()
+      .doc("Set the RocksDB merge operator version. This will be stored in the checkpoint when " +
+        "starting a streaming query. The checkpoint will use this merge operator version in the " +


nit: for the entire

anishshri-db · 2026-02-05T02:10:35Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala

  rocksDbOptions.setAvoidFlushDuringShutdown(true)
-  rocksDbOptions.setMergeOperator(new StringAppendOperator())
+  // Set merge operator based on version for backward compatibility
+  // Version 1: comma delimiter ",", Version 2: empty string ""


Should we note that this not documented but supported in RocksDB ?

anishshri-db · 2026-02-05T02:11:20Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala

+    case v => throw new IllegalArgumentException(
+      s"Invalid merge operator version: $v. Supported versions are 1 and 2")
+  }
+  rocksDbOptions.setMergeOperator(new StringAppendOperator(mergeDelimiter))


This is little different than using default constructor we were using earlier ? Will this be equivalent ?

It should be ',' vs "," and should be the same effect. There is a test confirming like merge("key1", "a") -> merge("key1", "b") -> get("key") with version 1 = "a,b" while with version 2 = "ab".

No I mean earlier we had this

rocksDbOptions.setMergeOperator(new StringAppendOperator())

but now we have this

rocksDbOptions.setMergeOperator(new StringAppendOperator(mergeDelimiter))

Would this be equivalent for the existing format ?

I haven't checked with JNI code of RocksDB but I think they are equivalent - I think we did not change the expected result for version 1 in the tests and the tests still pass.

anishshri-db · 2026-02-05T02:11:57Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala

+  // Version 2: Uses empty string "" as delimiter (no delimiter, direct concatenation)
+  //
+  // Note: this is also defined in `SQLConf.STATE_STORE_ROCKSDB_MERGE_OPERATOR_VERSION`.
+  // These two places should be updated together.


Why do we need this in 2 places ?

We need SQLConf to do the offset log trick. But in the meanwhile, it is easier for us to have the same entry here to automatically take the conf into account for RocksDBConf. We did this earlier - SQLConf.STATE_STORE_ROCKSDB_FORMAT_VERSION & FORMAT_VERSION.

anishshri-db · 2026-02-05T02:13:48Z

...re/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateStoreRowChecksum.scala


    // include the number of delimiters used for merge
-    resultSize += numValues - 1
+    resultSize += (numValues - 1) * delimiterSize


For empty delimiter, resultSize would be 0 then ?

It's just that we do not add the size for delimiters. resultSize being calculated in above logic retains the same.

anishshri-db · 2026-02-05T02:15:03Z

...e/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreSuite.scala

+        assert(valueRowToData(iterator0.next()) === 1)
+        assert(!iterator0.hasNext)
+
+        merge(store, "a", 0, 2)


Can we also add a test for some combination of put and merge followed by get as needed to verify the result for both merge delimiter versions ?

I'll add this tomorrow. This is the only review comment I'm yet to address.

959b0d6

Addressed.

…rator

…est case

anishshri-db · 2026-02-06T07:03:53Z

...e/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreSuite.scala

+
+  testMergeWithOperatorVersions(
+    "validate rocksdb values iterator correctness - put then merge") { _ =>
+


nit: extra newline ?

anishshri-db

lgtm pending nits and green CI

HeartSaVioR · 2026-02-06T12:53:08Z

The CI result of the commit f138772 was green.

https://github.com/apache/spark/runs/62718228501

The new commit after the commit only removed the new line which should be no-op.

88b369c

HeartSaVioR · 2026-02-06T12:53:39Z

Thanks! Merging to master.

…ksDB to empty string to concat without delimiter ### What changes were proposed in this pull request? This PR proposes to change the default delimiter for the merge operator of RocksDB to an empty string, so that merge operation does not add a delimiter and concat two without any character. Changing the delimiter isn't compatible with existing checkpoints, so this change is coupled with SQLConf, with known offset log metadata trick, to apply the change only for new streaming queries. * New SQL config: `spark.sql.streaming.stateStore.rocksdb.mergeOperatorVersion` * Default: 2 ('' as delimiter) * Default for existing checkpoints: 1 (',' as delimiter) ### Why are the changes needed? We found out there is no way to distinguish two cases of 1) put against non-existence value then merge and 2) merge against non-existence value then merge, from the current delimiter. There has been an "implication" that operators do ensure they call merge only when they know the operation is against existing key. This effectively requires GET operation which can be an outstanding performance impact depending on the logic. Making delimiter to an empty string (none) would eliminate the difference between the two cases, allowing operators to perform blind merge without checking the existence of the key. ### Does this PR introduce _any_ user-facing change? No, the change is internal and there is no user-facing change. ### How was this patch tested? Added UTs. ### Was this patch authored or co-authored using generative AI tooling? Co-authored by claude-4.5-sonnet Closes apache#54083 from HeartSaVioR/SPARK-55131. Authored-by: Jungtaek Lim <kabhwan.opensource@gmail.com> Signed-off-by: Jungtaek Lim <kabhwan.opensource@gmail.com>

github-actions bot added SQL STRUCTURED STREAMING labels Feb 1, 2026

HeartSaVioR mentioned this pull request Feb 1, 2026

[SPARK-55129][SS] Introduce new key encoders for timestamp as a first class (UnsafeRow) #53911

Closed

HeartSaVioR force-pushed the SPARK-55131 branch from bd195fe to aab929f Compare February 4, 2026 03:57

anishshri-db reviewed Feb 5, 2026

View reviewed changes

HeartSaVioR added 12 commits February 6, 2026 15:52

WIP - introduce a config to change the delimiter of RocksDB merge ope…

b077007

…rator

Reflect delimiter versions into multi-values logic

576ee86

enhance tests to cover both versions

b73d56b

cover test in RocksDBStateStore level

d27bc9c

revert refactor of the test and fit the new test case with existing t…

f4985f9

…est case

fix

770af0b

add tests for offset log

43454e5

silly fix

6b74c62

fix the bug

e902eba

fix test

e75023b

reflect review comment (not completed yet)

0f25e61

add new test

f138772

HeartSaVioR force-pushed the SPARK-55131 branch from 959b0d6 to f138772 Compare February 6, 2026 06:52

anishshri-db reviewed Feb 6, 2026

View reviewed changes

anishshri-db approved these changes Feb 6, 2026

View reviewed changes

nit fix

88b369c

HeartSaVioR closed this in b2601be Feb 6, 2026


		testMergeWithOperatorVersions(
		"validate rocksdb values iterator correctness - put then merge") { _ =>

Conversation

HeartSaVioR commented Feb 1, 2026

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

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

Uh oh!

github-actions bot commented Feb 1, 2026

JIRA Issue Information

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

anishshri-db left a comment

Choose a reason for hiding this comment

Uh oh!

HeartSaVioR commented Feb 6, 2026

Uh oh!

HeartSaVioR commented Feb 6, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants