Skip to content

[SPARK-56400][SS] Apply rangeScan API in transformWithState Timer/TTL#55265

Open
HeartSaVioR wants to merge 8 commits intoapache:masterfrom
HeartSaVioR:SPARK-56400-on-top-of-SPARK-56369
Open

[SPARK-56400][SS] Apply rangeScan API in transformWithState Timer/TTL#55265
HeartSaVioR wants to merge 8 commits intoapache:masterfrom
HeartSaVioR:SPARK-56400-on-top-of-SPARK-56369

Conversation

@HeartSaVioR
Copy link
Copy Markdown
Contributor

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

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
Copy link
Copy Markdown
Contributor Author

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

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant