[SPARK-56400][SS] Apply rangeScan API in transformWithState Timer/TTL by HeartSaVioR · Pull Request #55265 · apache/spark

HeartSaVioR · 2026-04-08T21:23:11Z

What changes were proposed in this pull request?

This PR proposes to apply rangeScan API in transformWithState Timer/TTL, which will give an improvement of scanning on expired timers and entries with configured TTL.

The main idea is to perform scanning of expired timers and TTL entries from [the end timestamp of previous scan + 1, new end timestamp], which was [None, new end timestamp]. Previously it had to go through tombstones prior batches made in prior evictions (till compaction happens), and with this change we will be able to skip those tombstones.

Why are the changes needed?

This change will give a hit to RocksDB about the exact range to scan, reducing the chance of reading tombstone a lot.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

New UTs, and existing UTs.

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

Generated-by: Claude 4.6 Opus

HeartSaVioR · 2026-04-08T21:23:43Z

Only the last commit is related to this PR. Once #55226 is merged, I'll rebase.

bogao007

Got some questions but LGTM overall!

bogao007 · 2026-04-10T00:03:04Z

  // The schema of the UnsafeRow returned by this iterator is (expirationMs, elementKey).
  private[sql] def ttlEvictionIterator(): Iterator[UnsafeRow] = {
-    val ttlIterator = store.iterator(TTL_INDEX)
+    val dummyElementKey = ELEMENT_KEY_PROJECTION


Is the dummyElementKey here just for using TTL_ENCODER.encodeTTLRow requires an elementKey?

Yes, correct.

bogao007 · 2026-04-10T00:04:10Z

+    } else {
+      None
+    }
+    val ttlIterator = store.rangeScan(startKey, endKey, TTL_INDEX)


Should we update the method comment indicating that we are using range scan now?

HeartSaVioR · 2026-04-16T01:10:32Z

cc. @viirya

Use bounded scan ranges in transformWithState TTL eviction and timer expiry to narrow the iteration scope: - TTLState.ttlEvictionIterator: use store.scan with startKey from prevBatchTimestampMs+1 and endKey from batchTimestampMs+1 to skip entries already cleaned up in the previous batch. - TimerStateImpl.getExpiredTimers: use store.scan with startKey from prevExpiryTimestampMs+1 and endKey from expiryTimestampMs+1. Processing-time timers use prevBatchTimestampMs; event-time timers use eventTimeWatermarkForLateEvents. Thread prevBatchTimestampMs from IncrementalExecution (via prevOffsetSeqMetadata) through TransformWithStateExec -> StatefulProcessorHandleImpl -> TTLState / TimerStateImpl. Copy UnsafeRow results from encodeTTLRow/UnsafeProjection to avoid the mutable-row-reuse bug where startKey and endKey alias the same internal buffer.

HeartSaVioR · 2026-04-18T13:19:50Z

@anishshri-db @viirya PTAL, thanks!

viirya · 2026-04-18T17:39:51Z

+      val row = new GenericInternalRow(keySchemaForSecIndex.length)
+      row.setLong(0, tsMs + 1)


Is it always valid to only fill the first field of keySchemaForSecIndex?

Unfortunately no - null is "greater" than non-null in UnsafeRow format. I'm going to apply the same approach with #55267 via using defaults on Literal.

viirya · 2026-04-18T17:47:08Z

+        val prevWatermark =
+          if (prevBatchTimestampMs.isDefined) eventTimeWatermarkForLateEvents else None


So if prevBatchTimestampMs is defined, the value of eventTimeWatermarkForLateEvents is always valid to use?

You spotted on.

This has some subtle bug - this does not work with legacy config (spark.sql.streaming.statefulOperator.allowMultiple=true). I think there is few users leveraging this config and we would probably want to kick the config out (we no longer need such a kill switch - this was introduced long time ago), so probably we can skip optimization for the legacy path, and remove the config sooner.

I indicated #55267 has the same issue, so I'm going to fix that PR as well.

HeartSaVioR · 2026-04-19T06:59:40Z

0c5e948

The code change is a big large - @viirya would you mind having another look and re-approve if the code change looks OK to you for your review comments? Thanks in advance!

viirya · 2026-04-19T07:13:50Z

+ * Caveats where the "smallest" property does NOT hold and which therefore should
+ * not appear in caller key schemas:
+ *   - `CharType(n)`: default is space-padded (0x20), but real values may legally
+ *     contain control bytes (0x00..0x1F).
+ *   - `VariantType`: the Variant binary layout is not guaranteed minimized by
+ *     `castToVariant(0, IntegerType)`.


If key schema includes these types, should we have some runtime detection?

Let's do that - I guess it's unlikely to hit but better to be safer.

291553b

I've taken a bit more work - we have a consistent binary lowest value for CharType, so we can actually address it without rejection. We can still reject VariantType since it's not practical for VariantType to be used in grouping key anyway.

The same change is applied to the PR here - #55267

@unstable

…chemas Address apache#55265 (comment): Literal.default's recursive "smallest" property does not hold for CharType or VariantType, so unguarded use can silently produce incorrect range-scan boundaries when such types appear in the key schema. - CharType(n): override with n zero-bytes (U+0000 repeated, UTF-8 encoded), which is the byte-wise smallest legitimate CharType(n) value. This keeps preserveCharVarcharTypeInfo=true users supported. - VariantType: assert-reject, because the Variant binary spec is @unstable and a hand-encoded minimum would be brittle. Existing Spark analysis already blocks VariantType in grouping / hashing positions, so this is defensive and should never fire in practice. Add RangeScanBoundaryUtilsSuite covering the happy path, byte-wise sanity check for CharType, nested CharType in struct, VarcharType empty default, and VariantType rejection at top level / nested in struct / nested in array.

@unstable

…chemas Address apache#55265 (comment): Literal.default's recursive "smallest" property does not hold for CharType or VariantType, so unguarded use can silently produce incorrect range-scan boundaries when such types appear in the key schema. - CharType(n): override with n zero-bytes (U+0000 repeated, UTF-8 encoded), which is the byte-wise smallest legitimate CharType(n) value. This keeps preserveCharVarcharTypeInfo=true users supported. - VariantType: assert-reject, because the Variant binary spec is @unstable and a hand-encoded minimum would be brittle. Existing Spark analysis already blocks VariantType in grouping / hashing positions, so this is defensive and should never fire in practice. Add RangeScanBoundaryUtilsSuite covering the happy path, byte-wise sanity check for CharType, nested CharType in struct, VarcharType empty default, and VariantType rejection at top level / nested in struct / nested in array.

HeartSaVioR · 2026-04-19T08:07:28Z

It's unfortunate that null field is flipped as a binary order in UnsafeRow so we are doing a heavy lifting work. (Otherwise this should be just filling nulls.) Also technically we should just ask the caller to fill out the range part of the data on calling rangeScan, not to fill out all keys. Arguably if we are very clear about the intention of the scan e.g. timestamp in API, we really don't need to do the heavy work like this.

HeartSaVioR · 2026-04-19T21:10:54Z

https://github.com/HeartSaVioR/spark/actions/runs/24624331383/job/72002194205

CI only fails on k8s integratino test and sparkr which are irrelevant.

HeartSaVioR · 2026-04-19T21:13:21Z

Thanks! Merging to master.

@unstable

…chemas Address apache#55265 (comment): Literal.default's recursive "smallest" property does not hold for CharType or VariantType, so unguarded use can silently produce incorrect range-scan boundaries when such types appear in the key schema. - CharType(n): override with n zero-bytes (U+0000 repeated, UTF-8 encoded), which is the byte-wise smallest legitimate CharType(n) value. This keeps preserveCharVarcharTypeInfo=true users supported. - VariantType: assert-reject, because the Variant binary spec is @unstable and a hand-encoded minimum would be brittle. Existing Spark analysis already blocks VariantType in grouping / hashing positions, so this is defensive and should never fire in practice. Add RangeScanBoundaryUtilsSuite covering the happy path, byte-wise sanity check for CharType, nested CharType in struct, VarcharType empty default, and VariantType rejection at top level / nested in struct / nested in array.

HeartSaVioR mentioned this pull request Apr 8, 2026

[WIP][REVIEW-DONE] Apply StateStore's scan operation to optimize the pattern of range scan for TWS HeartSaVioR/spark#25

Closed

bogao007 approved these changes Apr 10, 2026

View reviewed changes

HeartSaVioR force-pushed the SPARK-56400-on-top-of-SPARK-56369 branch from 3c99f65 to 6025cf0 Compare April 15, 2026 03:34

HeartSaVioR added 3 commits April 18, 2026 21:53

Reflect review comment

7da3f96

empty commit to retrigger CI

5520bb6

HeartSaVioR force-pushed the SPARK-56400-on-top-of-SPARK-56369 branch from 6025cf0 to 5520bb6 Compare April 18, 2026 12:54

viirya approved these changes Apr 18, 2026

View reviewed changes

viirya reviewed Apr 18, 2026

View reviewed changes

anishshri-db approved these changes Apr 18, 2026

View reviewed changes

viirya reviewed Apr 18, 2026

View reviewed changes

Review comments

0c5e948

HeartSaVioR requested a review from viirya April 19, 2026 06:59

viirya reviewed Apr 19, 2026

View reviewed changes

HeartSaVioR added a commit to HeartSaVioR/spark that referenced this pull request Apr 19, 2026

Fix the bug found in apache#55265 (comment)

4b36621

HeartSaVioR requested a review from viirya April 19, 2026 08:02

viirya approved these changes Apr 19, 2026

View reviewed changes

HeartSaVioR closed this in f151e03 Apr 19, 2026

HeartSaVioR mentioned this pull request Apr 19, 2026

[SPARK-56402][SS] Apply rangeScan API in stream-stream join format version 4 #55267

Closed

HeartSaVioR added a commit to HeartSaVioR/spark that referenced this pull request Apr 19, 2026

Fix the bug found in apache#55265 (comment)

d91de33

HeartSaVioR added a commit to HeartSaVioR/spark that referenced this pull request Apr 19, 2026

Fix the bug found in apache#55265 (comment)

9e41e3f

		val row = new GenericInternalRow(keySchemaForSecIndex.length)
		row.setLong(0, tsMs + 1)

		val prevWatermark =
		if (prevBatchTimestampMs.isDefined) eventTimeWatermarkForLateEvents else None

Conversation

HeartSaVioR commented Apr 8, 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!

HeartSaVioR commented Apr 8, 2026

Uh oh!

bogao007 left a comment

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!

HeartSaVioR commented Apr 16, 2026

Uh oh!

HeartSaVioR commented Apr 18, 2026

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!

HeartSaVioR commented Apr 19, 2026

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!

HeartSaVioR commented Apr 19, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

HeartSaVioR commented Apr 19, 2026

Uh oh!

HeartSaVioR commented Apr 19, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

HeartSaVioR commented Apr 19, 2026 •

edited

Loading