You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@spark.apache.org by "Jungtaek Lim (Jira)" <ji...@apache.org> on 2024/11/26 23:30:00 UTC
[jira] [Resolved] (SPARK-50302) TransformWithState secondary index sizes should be proportional to primary indexes
[ https://issues.apache.org/jira/browse/SPARK-50302?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Jungtaek Lim resolved SPARK-50302.
----------------------------------
Fix Version/s: 4.0.0
Resolution: Fixed
Issue resolved by pull request 48853
[https://github.com/apache/spark/pull/48853]
> TransformWithState secondary index sizes should be proportional to primary indexes
> ----------------------------------------------------------------------------------
>
> Key: SPARK-50302
> URL: https://issues.apache.org/jira/browse/SPARK-50302
> Project: Spark
> Issue Type: Improvement
> Components: Structured Streaming
> Affects Versions: 4.0.0
> Reporter: Neil Ramaswamy
> Assignee: Neil Ramaswamy
> Priority: Major
> Labels: pull-request-available
> Fix For: 4.0.0
>
>
> Currently, the TWS operator handles TTL state variables in the same approximate way:
> # Upsert the value into the primary index
> # Upsert the expiration timestamp into the secondary index, where the expiration timestamp is the `batchTimestampMs`
> The issue with this approach is that if the same state variable is updated across two different micro-batches, there exists 1 entry in the primary index, while there exists _two_ entries in the secondary index. Consider the following example for a state variable `foo` with value `v1`, and TTL delay of 500:
> Batch 0, `batchTimestampMs = 100`, `foo` updates to `v1`:
> * Primary index: `[foo -> (v1, 600)]`
> * Secondary index: `[(600, foo) -> EMPTY]`
> Batch 1: `batchTimestampMs = 200`, `foo` updates to `v2`:
> * Primary index: `[foo -> (v2, 700)]`
> * Secondary index: `[(600, foo) -> EMPTY, (700, foo) -> EMPTY]`
> You'll notice that the secondary index now has size 2, even though the primary index only has size 1. When we clean up `(600, foo)`, we actually _don't_ delete it, since we do [another lookup in the primary index to determine if the secondary index entry we're dealing with is stale|https://github.com/apache/spark/blob/05508cf7cb9da3042fa4b17645102a6406278695/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/ValueStateImplWithTTL.scala#L109-L113].
> So, to summarize, the write path is always write to the primary _and_ secondary index, and then for deletion, we go through everything in the secondary index, and do a lookup on the primary index to see whether to delete it.
> While this may not seem like a huge issue, things get way worse for `ListState`. Our cleanup logic for list state is as follows:
> # Grab an iterator for the entire list
> # Clear the entire list
> # For each element in the iterator that isn't expired, merge it back into the list
> This means that having an erroneous entry in your secondary index means that you will go through the entire list _several_ times, which will negatively impact performance. We should most definitely make sure that the secondary index has only as many elements as the primary index, which will prevent us from doing unnecessary work during cleanup.
> Solutions to this problem will be proposed in the PR.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org