You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "ericm-db (via GitHub)" <gi...@apache.org> on 2025/04/28 17:24:05 UTC
[PR] [WIP] Adding release() to ReadStateStore interface [spark]
ericm-db opened a new pull request, #50742:
URL: https://github.com/apache/spark/pull/50742
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html
2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][SPARK-XXXX] Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a faster review.
7. If you want to add a new configuration, please read the guideline first for naming configurations in
'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
8. If you want to add or modify an error type or message, please read the guideline first in
'common/utils/src/main/resources/error/README.md'.
-->
### What changes were proposed in this pull request?
<!--
Please clarify what changes you are proposing. The purpose of this section is to outline the changes and how this PR fixes the issue.
If possible, please consider writing useful notes for better and faster reviews in your PR. See the examples below.
1. If you refactor some codes with changing classes, showing the class hierarchy will help reviewers.
2. If you fix some SQL features, you can provide some references of other DBMSes.
3. If there is design documentation, please add the link.
4. If there is a discussion in the mailing list, please add the link.
-->
### Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
1. If you propose a new API, clarify the use case for a new API.
2. If you fix a bug, you can clarify why it is a bug.
-->
### Does this PR introduce _any_ user-facing change?
<!--
Note that it means *any* user-facing change including all aspects such as new features, bug fixes, or other behavior changes. Documentation-only updates are not considered user-facing changes.
If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description and/or an example to show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change compared to the released Spark versions or within the unreleased branches such as master.
If no, write 'No'.
-->
### How was this patch tested?
<!--
If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible.
If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future.
If tests were not added, please describe why they were not added and/or why it was difficult to add.
If benchmark tests were added, please run the benchmarks in GitHub Actions for the consistent environment, and the instructions could accord to: https://spark.apache.org/developer-tools.html#github-workflow-benchmarks.
-->
### Was this patch authored or co-authored using generative AI tooling?
<!--
If generative AI tooling has been used in the process of authoring this patch, please include the
phrase: 'Generated-by: ' followed by the name of the tool and its version.
If no, write 'No'.
Please refer to the [ASF Generative Tooling Guidance](https://www.apache.org/legal/generative-tooling.html) for details.
-->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51955] Adding release() to ReadStateStore interface and reusing ReadStore for Streaming Aggregations [spark]
Posted by "liviazhu-db (via GitHub)" <gi...@apache.org>.
liviazhu-db commented on code in PR #50742:
URL: https://github.com/apache/spark/pull/50742#discussion_r2067532717
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateStoreRDD.scala:
##########
@@ -82,19 +106,47 @@ class ReadStateStoreRDD[T: ClassTag, U: ClassTag](
useColumnFamilies: Boolean = false,
extraOptions: Map[String, String] = Map.empty)
extends BaseStateStoreRDD[T, U](dataRDD, checkpointLocation, queryRunId, operatorId,
- sessionState, storeCoordinator, extraOptions) {
+ sessionState, storeCoordinator, extraOptions) with StateStoreRDDProvider {
+
+ // Using a ConcurrentHashMap to track state stores by partition ID
+ // and whether this store was used to create a write store or not.
+ @transient private lazy val partitionStores =
Review Comment:
Will there ever be more than 1 store added here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51955] Adding release() to ReadStateStore interface and reusing ReadStore for Streaming Aggregations [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on code in PR #50742:
URL: https://github.com/apache/spark/pull/50742#discussion_r2067620345
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateStore.scala:
##########
@@ -565,6 +582,29 @@ trait StateStoreProvider {
version: Long,
stateStoreCkptId: Option[String] = None): StateStore
+ /**
+ * Creates a writable store from an existing read-only store for the specified version.
+ *
+ * This method enables an important optimization pattern for stateful operations where
+ * the same state store needs to be accessed for both reading and writing within a task.
+ * Instead of opening two separate state store instances (which can cause contention issues),
+ * this method converts an existing read-only store to a writable store that can commit changes.
+ *
+ * This approach is particularly beneficial when:
+ * - A stateful operation needs to first read the existing state, then update it
+ * - The state store has locking mechanisms that prevent concurrent access
+ * - Multiple state store connections would cause unnecessary resource duplication
+ *
+ * @param readStore The existing read-only store instance to convert to a writable store
+ * @param version The version of the state store (must match the read store's version)
+ * @param uniqueId Optional unique identifier for checkpointing
+ * @return A writable StateStore instance that can be used to update and commit changes
+ */
+ def getWriteStoreFromReadStore(
Review Comment:
ni: should we rename as `upgradeReadStoreToWriteStore` ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51955] Adding release() to ReadStateStore interface and reusing ReadStore for Streaming Aggregations [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on PR #50742:
URL: https://github.com/apache/spark/pull/50742#issuecomment-2840548793
cc - @cloud-fan - could you PTAL too ? especially around the RDD interactions ? Thx
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51955] Adding release() to ReadStateStore interface and reusing ReadStore for Streaming Aggregations [spark]
Posted by "ericm-db (via GitHub)" <gi...@apache.org>.
ericm-db commented on code in PR #50742:
URL: https://github.com/apache/spark/pull/50742#discussion_r2070607523
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala:
##########
@@ -446,17 +459,48 @@ private[sql] class RocksDBStateStoreProvider
override def stateStoreId: StateStoreId = stateStoreId_
- override def getStore(version: Long, uniqueId: Option[String] = None): StateStore = {
+ /**
+ * Creates and returns a state store with the specified parameters.
+ *
+ * @param version The version of the state store to load
+ * @param uniqueId Optional unique identifier for checkpoint
+ * @param readOnly Whether to open the store in read-only mode
+ * @param existingStore Optional existing store to reuse instead of creating a new one
+ * @return The loaded state store
+ */
+ private def loadStateStore(
+ version: Long,
+ uniqueId: Option[String],
+ readOnly: Boolean,
+ existingStore: Option[ReadStateStore] = None): StateStore = {
try {
if (version < 0) {
throw QueryExecutionErrors.unexpectedStateStoreVersion(version)
}
- rocksDB.load(
- version,
- stateStoreCkptId = if (storeConf.enableStateStoreCheckpointIds) uniqueId else None)
- new RocksDBStateStore(version)
- }
- catch {
+ try {
+ // Load RocksDB store
+ rocksDB.load(
+ version,
+ stateStoreCkptId = if (storeConf.enableStateStoreCheckpointIds) uniqueId else None,
+ readOnly = readOnly)
+
+ // Return appropriate store instance
+ existingStore match {
+ case Some(stateStore: RocksDBStateStore) =>
+ // Reuse existing store for getWriteStore case
+ StateStoreThreadLocalTracker.setUsedForWriteStore(true)
Review Comment:
Yeah I wanted to - but I think there would be some abstraction leak since we don't 'reuse' stores for HDFS, we just open a new one.
The reason for this is because HDFSBackedStateStore and HDFSBackedReadStateStore are different classes, so we can't directly return the store like we do [here](https://github.com/apache/spark/pull/50742/files#diff-76a55d7a6f0b91a19366e01c919c615a4d13afc1c1c37b017afa9df1c1705218R492)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51955] Adding release() to ReadStateStore interface and reusing ReadStore for Streaming Aggregations [spark]
Posted by "ericm-db (via GitHub)" <gi...@apache.org>.
ericm-db commented on code in PR #50742:
URL: https://github.com/apache/spark/pull/50742#discussion_r2067627565
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateStoreRDD.scala:
##########
@@ -82,19 +107,47 @@ class ReadStateStoreRDD[T: ClassTag, U: ClassTag](
useColumnFamilies: Boolean = false,
extraOptions: Map[String, String] = Map.empty)
extends BaseStateStoreRDD[T, U](dataRDD, checkpointLocation, queryRunId, operatorId,
- sessionState, storeCoordinator, extraOptions) {
+ sessionState, storeCoordinator, extraOptions) with StateStoreRDDProvider {
+
+ // Using a ConcurrentHashMap to track state stores by partition ID
+ // and whether this store was used to create a write store or not.
+ @transient private lazy val partitionStores =
+ new java.util.concurrent.ConcurrentHashMap[StateStoreProviderId, (ReadStateStore, Boolean)]()
+
+ override def getStateStoreForPartition(
+ providerId: StateStoreProviderId): Option[ReadStateStore] = {
+ val (readStore, _) = partitionStores.get(providerId)
+ partitionStores.put(providerId, (readStore, true))
+ Option(readStore)
+ }
override protected def getPartitions: Array[Partition] = dataRDD.partitions
override def compute(partition: Partition, ctxt: TaskContext): Iterator[U] = {
val storeProviderId = getStateProviderId(partition)
+ val partitionId = partition.index
val inputIter = dataRDD.iterator(partition, ctxt)
val store = StateStore.getReadOnly(
storeProviderId, keySchema, valueSchema, keyStateEncoderSpec, storeVersion,
- stateStoreCkptIds.map(_.apply(partition.index).head),
+ stateStoreCkptIds.map(_.apply(partitionId).head),
Review Comment:
No I think these are equivalent though. I'll change it back
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51955] Adding release() to ReadStateStore interface and reusing ReadStore for Streaming Aggregations [spark]
Posted by "ericm-db (via GitHub)" <gi...@apache.org>.
ericm-db commented on code in PR #50742:
URL: https://github.com/apache/spark/pull/50742#discussion_r2070624489
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala:
##########
@@ -446,17 +459,48 @@ private[sql] class RocksDBStateStoreProvider
override def stateStoreId: StateStoreId = stateStoreId_
- override def getStore(version: Long, uniqueId: Option[String] = None): StateStore = {
+ /**
+ * Creates and returns a state store with the specified parameters.
+ *
+ * @param version The version of the state store to load
+ * @param uniqueId Optional unique identifier for checkpoint
+ * @param readOnly Whether to open the store in read-only mode
+ * @param existingStore Optional existing store to reuse instead of creating a new one
+ * @return The loaded state store
+ */
+ private def loadStateStore(
+ version: Long,
+ uniqueId: Option[String],
+ readOnly: Boolean,
+ existingStore: Option[ReadStateStore] = None): StateStore = {
try {
if (version < 0) {
throw QueryExecutionErrors.unexpectedStateStoreVersion(version)
}
- rocksDB.load(
- version,
- stateStoreCkptId = if (storeConf.enableStateStoreCheckpointIds) uniqueId else None)
- new RocksDBStateStore(version)
- }
- catch {
+ try {
+ // Load RocksDB store
+ rocksDB.load(
+ version,
+ stateStoreCkptId = if (storeConf.enableStateStoreCheckpointIds) uniqueId else None,
+ readOnly = readOnly)
+
+ // Return appropriate store instance
+ existingStore match {
+ case Some(stateStore: RocksDBStateStore) =>
+ // Reuse existing store for getWriteStore case
+ StateStoreThreadLocalTracker.setUsedForWriteStore(true)
Review Comment:
Thanks for the suggestion, it worked.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51955] Adding release() to ReadStateStore interface and reusing ReadStore for Streaming Aggregations [spark]
Posted by "ericm-db (via GitHub)" <gi...@apache.org>.
ericm-db commented on code in PR #50742:
URL: https://github.com/apache/spark/pull/50742#discussion_r2070674011
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateStoreRDD.scala:
##########
@@ -27,6 +27,28 @@ import org.apache.spark.sql.internal.SessionState
import org.apache.spark.sql.types.StructType
import org.apache.spark.util.SerializableConfiguration
+/**
+ * Thread local storage for sharing StateStore instances between RDDs.
+ * This allows a ReadStateStore to be reused by a subsequent StateStore operation.
+ */
+object StateStoreThreadLocalTracker {
+ private val readStore: ThreadLocal[ReadStateStore] = new ThreadLocal[ReadStateStore]
Review Comment:
Sure yeah
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51955] Adding release() to ReadStateStore interface and reusing ReadStore for Streaming Aggregations [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on code in PR #50742:
URL: https://github.com/apache/spark/pull/50742#discussion_r2070681239
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateStoreRDD.scala:
##########
@@ -27,6 +27,43 @@ import org.apache.spark.sql.internal.SessionState
import org.apache.spark.sql.types.StructType
import org.apache.spark.util.SerializableConfiguration
+/**
+ * Thread local storage for sharing StateStore instances between RDDs.
+ * This allows a ReadStateStore to be reused by a subsequent StateStore operation.
+ */
+object StateStoreThreadLocalTracker {
+ /** Case class to hold both the store and its usage state */
+ case class StoreInfo(store: ReadStateStore, usedForWriteStore: Boolean = false)
Review Comment:
nit: maybe move members to a new line each ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51955] Adding release() to ReadStateStore interface and reusing ReadStore for Streaming Aggregations [spark]
Posted by "liviazhu-db (via GitHub)" <gi...@apache.org>.
liviazhu-db commented on PR #50742:
URL: https://github.com/apache/spark/pull/50742#issuecomment-2840484430
@ericm-db Can you update the PR description to be more specific about the inefficiency? Basically that in the current impl, we always abort read store, triggering unnecessary reload of the state store.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51955] Adding release() to ReadStateStore interface and reusing ReadStore for Streaming Aggregations [spark]
Posted by "liviazhu-db (via GitHub)" <gi...@apache.org>.
liviazhu-db commented on code in PR #50742:
URL: https://github.com/apache/spark/pull/50742#discussion_r2067517342
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala:
##########
@@ -468,29 +511,20 @@ private[sql] class RocksDBStateStoreProvider
case e: Throwable => throw QueryExecutionErrors.cannotLoadStore(e)
}
}
+ override def getStore(version: Long, uniqueId: Option[String] = None): StateStore = {
+ loadStateStore(version, uniqueId, readOnly = false)
+ }
+
+ override def getWriteStore(
Review Comment:
nit: rename to getWriteStoreFromReadStore?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateStore.scala:
##########
@@ -565,6 +582,11 @@ trait StateStoreProvider {
version: Long,
stateStoreCkptId: Option[String] = None): StateStore
+ def getWriteStore(
Review Comment:
add docs comment?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateStoreRDD.scala:
##########
@@ -82,19 +106,47 @@ class ReadStateStoreRDD[T: ClassTag, U: ClassTag](
useColumnFamilies: Boolean = false,
extraOptions: Map[String, String] = Map.empty)
extends BaseStateStoreRDD[T, U](dataRDD, checkpointLocation, queryRunId, operatorId,
- sessionState, storeCoordinator, extraOptions) {
+ sessionState, storeCoordinator, extraOptions) with StateStoreRDDProvider {
+
+ // Using a ConcurrentHashMap to track state stores by partition ID
+ // and whether this store was used to create a write store or not.
+ @transient private lazy val partitionStores =
+ new java.util.concurrent.ConcurrentHashMap[Int, (ReadStateStore, Boolean)]()
+
+ override def getStateStoreForPartition(partitionId: Int): Option[ReadStateStore] = {
+ val (readStore, _) = partitionStores.get(partitionId)
+ partitionStores.put(partitionId, (readStore, true))
+ Option(readStore)
+ }
override protected def getPartitions: Array[Partition] = dataRDD.partitions
override def compute(partition: Partition, ctxt: TaskContext): Iterator[U] = {
val storeProviderId = getStateProviderId(partition)
+ val partitionId = partition.index
val inputIter = dataRDD.iterator(partition, ctxt)
val store = StateStore.getReadOnly(
storeProviderId, keySchema, valueSchema, keyStateEncoderSpec, storeVersion,
- stateStoreCkptIds.map(_.apply(partition.index).head),
+ stateStoreCkptIds.map(_.apply(partitionId).head),
stateSchemaBroadcast,
useColumnFamilies, storeConf, hadoopConfBroadcast.value.value)
+
+ // Store reference for this partition
+ partitionStores.put(partitionId, (store, false))
+
+ // Register a cleanup callback to be executed when the task completes
+ ctxt.addTaskCompletionListener[Unit](_ => {
Review Comment:
Why are we adding the listeners here? Is it different from the one in mapPartitionsWithReadStateStore?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateStoreRDD.scala:
##########
@@ -82,19 +106,47 @@ class ReadStateStoreRDD[T: ClassTag, U: ClassTag](
useColumnFamilies: Boolean = false,
extraOptions: Map[String, String] = Map.empty)
extends BaseStateStoreRDD[T, U](dataRDD, checkpointLocation, queryRunId, operatorId,
- sessionState, storeCoordinator, extraOptions) {
+ sessionState, storeCoordinator, extraOptions) with StateStoreRDDProvider {
+
+ // Using a ConcurrentHashMap to track state stores by partition ID
+ // and whether this store was used to create a write store or not.
+ @transient private lazy val partitionStores =
Review Comment:
Will there ever be more than 1 store added here?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala:
##########
@@ -468,29 +511,20 @@ private[sql] class RocksDBStateStoreProvider
case e: Throwable => throw QueryExecutionErrors.cannotLoadStore(e)
}
}
+ override def getStore(version: Long, uniqueId: Option[String] = None): StateStore = {
+ loadStateStore(version, uniqueId, readOnly = false)
+ }
+
+ override def getWriteStore(
+ readStore: ReadStateStore,
+ version: Long,
+ uniqueId: Option[String] = None): StateStore = {
+ assert(version == readStore.version)
Review Comment:
Can you leave a comment or more informative error msg here?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateStore.scala:
##########
@@ -950,6 +972,30 @@ object StateStore extends Logging {
storeProvider.getReadStore(version, stateStoreCkptId)
}
+ def getWriteStore(
+ readStore: ReadStateStore,
+ storeProviderId: StateStoreProviderId,
+ keySchema: StructType,
+ valueSchema: StructType,
+ keyStateEncoderSpec: KeyStateEncoderSpec,
+ version: Long,
+ stateStoreCkptId: Option[String],
+ stateSchemaBroadcast: Option[StateSchemaBroadcast],
+ useColumnFamilies: Boolean,
+ storeConf: StateStoreConf,
+ hadoopConf: Configuration,
+ useMultipleValuesPerKey: Boolean = false): StateStore = {
+ hadoopConf.set(StreamExecution.RUN_ID_KEY, storeProviderId.queryRunId.toString)
+ if (version < 0) {
+ throw QueryExecutionErrors.unexpectedStateStoreVersion(version)
+ }
+ hadoopConf.set(StreamExecution.RUN_ID_KEY, storeProviderId.queryRunId.toString)
Review Comment:
Why are we setting this twice? Can you add more comments about what is going on here
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51955] Adding release() to ReadStateStore interface and reusing ReadStore for Streaming Aggregations [spark]
Posted by "micheal-o (via GitHub)" <gi...@apache.org>.
micheal-o commented on code in PR #50742:
URL: https://github.com/apache/spark/pull/50742#discussion_r2078913287
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/HDFSBackedStateStoreProvider.scala:
##########
@@ -89,6 +89,8 @@ private[sql] class HDFSBackedStateStoreProvider extends StateStoreProvider with
override def abort(): Unit = {}
Review Comment:
I thought we are removing `abort` api in readstore?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala:
##########
@@ -365,6 +366,18 @@ private[sql] class RocksDBStateStoreProvider
}
result
}
+
+ override def release(): Unit = {
+ if (state != RELEASED) {
Review Comment:
ditto, this should not be allowed on statestore
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala:
##########
@@ -446,17 +459,49 @@ private[sql] class RocksDBStateStoreProvider
override def stateStoreId: StateStoreId = stateStoreId_
- override def getStore(version: Long, uniqueId: Option[String] = None): StateStore = {
+ /**
+ * Creates and returns a state store with the specified parameters.
+ *
+ * @param version The version of the state store to load
+ * @param uniqueId Optional unique identifier for checkpoint
+ * @param readOnly Whether to open the store in read-only mode
+ * @param existingStore Optional existing store to reuse instead of creating a new one
+ * @return The loaded state store
+ */
+ private def loadStateStore(
+ version: Long,
+ uniqueId: Option[String],
+ readOnly: Boolean,
+ existingStore: Option[ReadStateStore] = None): StateStore = {
try {
if (version < 0) {
throw QueryExecutionErrors.unexpectedStateStoreVersion(version)
}
- rocksDB.load(
- version,
- stateStoreCkptId = if (storeConf.enableStateStoreCheckpointIds) uniqueId else None)
- new RocksDBStateStore(version)
- }
- catch {
+ try {
+ // Load RocksDB store
+ rocksDB.load(
+ version,
+ stateStoreCkptId = if (storeConf.enableStateStoreCheckpointIds) uniqueId else None,
+ readOnly = readOnly)
+
+ // Return appropriate store instance
+ existingStore match {
+ // We need to match like this as opposed to case Some(ss: RocksDBStateStore)
+ // because of how the tests create the class in StateStoreRDDSuite
+ case Some(stateStore: ReadStateStore) if stateStore.isInstanceOf[RocksDBStateStore] =>
+ stateStore.asInstanceOf[StateStore]
+ case Some(other) =>
+ throw new IllegalArgumentException(s"Existing store must be a RocksDBStateStore," +
+ s" store is actually ${other.getClass.getSimpleName}")
+ case None =>
+ // Create new store instance for getStore/getReadStore cases
+ new RocksDBStateStore(version)
+ }
+ } catch {
+ case e: Throwable =>
Review Comment:
why are we doing this?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/package.scala:
##########
@@ -109,8 +116,12 @@ package object state {
val cleanedF = dataRDD.sparkContext.clean(storeReadFn)
val wrappedF = (store: ReadStateStore, iter: Iterator[T]) => {
// Clean up the state store.
- TaskContext.get().addTaskCompletionListener[Unit](_ => {
- store.abort()
+ val ctxt = TaskContext.get()
+ ctxt.addTaskCompletionListener[Unit](_ => {
+ if (!StateStoreThreadLocalTracker.isUsedForWriteStore) {
Review Comment:
add comment
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala:
##########
@@ -446,17 +459,49 @@ private[sql] class RocksDBStateStoreProvider
override def stateStoreId: StateStoreId = stateStoreId_
- override def getStore(version: Long, uniqueId: Option[String] = None): StateStore = {
+ /**
+ * Creates and returns a state store with the specified parameters.
+ *
+ * @param version The version of the state store to load
+ * @param uniqueId Optional unique identifier for checkpoint
+ * @param readOnly Whether to open the store in read-only mode
+ * @param existingStore Optional existing store to reuse instead of creating a new one
+ * @return The loaded state store
+ */
+ private def loadStateStore(
+ version: Long,
+ uniqueId: Option[String],
+ readOnly: Boolean,
+ existingStore: Option[ReadStateStore] = None): StateStore = {
try {
if (version < 0) {
throw QueryExecutionErrors.unexpectedStateStoreVersion(version)
}
- rocksDB.load(
- version,
- stateStoreCkptId = if (storeConf.enableStateStoreCheckpointIds) uniqueId else None)
- new RocksDBStateStore(version)
- }
- catch {
+ try {
+ // Load RocksDB store
+ rocksDB.load(
+ version,
+ stateStoreCkptId = if (storeConf.enableStateStoreCheckpointIds) uniqueId else None,
+ readOnly = readOnly)
+
+ // Return appropriate store instance
+ existingStore match {
+ // We need to match like this as opposed to case Some(ss: RocksDBStateStore)
+ // because of how the tests create the class in StateStoreRDDSuite
+ case Some(stateStore: ReadStateStore) if stateStore.isInstanceOf[RocksDBStateStore] =>
Review Comment:
this check isn't sufficient because it would succeed if I pass in a RocksDBStateStore that wasn't created as readonly. Instead we should check if the RocksDBStateStore passed in was created as readonly?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala:
##########
@@ -365,6 +366,18 @@ private[sql] class RocksDBStateStoreProvider
}
result
}
+
+ override def release(): Unit = {
+ if (state != RELEASED) {
+ logInfo(log"Releasing ${MDC(VERSION_NUM, version + 1)} " +
Review Comment:
why is it version + 1 here?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/HDFSBackedStateStoreProvider.scala:
##########
@@ -194,6 +197,10 @@ private[sql] class HDFSBackedStateStoreProvider extends StateStoreProvider with
log"for ${MDC(LogKeys.STATE_STORE_PROVIDER, this)}")
}
+ override def release(): Unit = {
Review Comment:
release should only be allowed on readstore and shouldn't be allowed on statestore. State store should only allow `abort` or `commit`
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala:
##########
@@ -446,17 +459,49 @@ private[sql] class RocksDBStateStoreProvider
override def stateStoreId: StateStoreId = stateStoreId_
- override def getStore(version: Long, uniqueId: Option[String] = None): StateStore = {
+ /**
+ * Creates and returns a state store with the specified parameters.
+ *
+ * @param version The version of the state store to load
+ * @param uniqueId Optional unique identifier for checkpoint
+ * @param readOnly Whether to open the store in read-only mode
+ * @param existingStore Optional existing store to reuse instead of creating a new one
+ * @return The loaded state store
+ */
+ private def loadStateStore(
+ version: Long,
+ uniqueId: Option[String],
+ readOnly: Boolean,
+ existingStore: Option[ReadStateStore] = None): StateStore = {
try {
if (version < 0) {
throw QueryExecutionErrors.unexpectedStateStoreVersion(version)
}
- rocksDB.load(
- version,
- stateStoreCkptId = if (storeConf.enableStateStoreCheckpointIds) uniqueId else None)
- new RocksDBStateStore(version)
- }
- catch {
+ try {
+ // Load RocksDB store
+ rocksDB.load(
+ version,
+ stateStoreCkptId = if (storeConf.enableStateStoreCheckpointIds) uniqueId else None,
+ readOnly = readOnly)
+
+ // Return appropriate store instance
+ existingStore match {
+ // We need to match like this as opposed to case Some(ss: RocksDBStateStore)
+ // because of how the tests create the class in StateStoreRDDSuite
+ case Some(stateStore: ReadStateStore) if stateStore.isInstanceOf[RocksDBStateStore] =>
+ stateStore.asInstanceOf[StateStore]
+ case Some(other) =>
+ throw new IllegalArgumentException(s"Existing store must be a RocksDBStateStore," +
Review Comment:
Should we throw error earlier if a bad store was passed in before we do load?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/package.scala:
##########
@@ -67,9 +68,15 @@ package object state {
val cleanedF = dataRDD.sparkContext.clean(storeUpdateFunction)
val wrappedF = (store: StateStore, iter: Iterator[T]) => {
// Abort the state store in case of error
- TaskContext.get().addTaskCompletionListener[Unit](_ => {
+ val ctxt = TaskContext.get()
+ ctxt.addTaskCompletionListener[Unit](_ => {
if (!store.hasCommitted) store.abort()
})
+ ctxt.addTaskFailureListener(new TaskFailureListener {
Review Comment:
why the failure listener?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/HDFSBackedStateStoreProvider.scala:
##########
@@ -89,6 +89,8 @@ private[sql] class HDFSBackedStateStoreProvider extends StateStoreProvider with
override def abort(): Unit = {}
Review Comment:
ok, I think I remember a conversation we had about leaving it for now. But with this new change, can you check if there is still any usage of calling abort on readstore?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateStore.scala:
##########
@@ -950,6 +989,29 @@ object StateStore extends Logging {
storeProvider.getReadStore(version, stateStoreCkptId)
}
+ def getWriteStore(
Review Comment:
nit: add func comment
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51955] Adding release() to ReadStateStore interface and reusing ReadStore for Streaming Aggregations [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on code in PR #50742:
URL: https://github.com/apache/spark/pull/50742#discussion_r2067623127
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateStoreRDD.scala:
##########
@@ -34,7 +59,7 @@ abstract class BaseStateStoreRDD[T: ClassTag, U: ClassTag](
operatorId: Long,
sessionState: SessionState,
@transient private val storeCoordinator: Option[StateStoreCoordinatorRef],
- extraOptions: Map[String, String] = Map.empty) extends RDD[U](dataRDD) {
+ extraOptions: Map[String, String] = Map.empty) extends RDD[U](dataRDD) with Logging {
Review Comment:
intentional ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51955] Adding release() to ReadStateStore interface and reusing ReadStore for Streaming Aggregations [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on code in PR #50742:
URL: https://github.com/apache/spark/pull/50742#discussion_r2067603590
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateStore.scala:
##########
@@ -950,6 +972,30 @@ object StateStore extends Logging {
storeProvider.getReadStore(version, stateStoreCkptId)
}
+ def getWriteStore(
+ readStore: ReadStateStore,
+ storeProviderId: StateStoreProviderId,
+ keySchema: StructType,
+ valueSchema: StructType,
+ keyStateEncoderSpec: KeyStateEncoderSpec,
+ version: Long,
+ stateStoreCkptId: Option[String],
+ stateSchemaBroadcast: Option[StateSchemaBroadcast],
+ useColumnFamilies: Boolean,
+ storeConf: StateStoreConf,
+ hadoopConf: Configuration,
+ useMultipleValuesPerKey: Boolean = false): StateStore = {
+ hadoopConf.set(StreamExecution.RUN_ID_KEY, storeProviderId.queryRunId.toString)
+ if (version < 0) {
+ throw QueryExecutionErrors.unexpectedStateStoreVersion(version)
+ }
+ hadoopConf.set(StreamExecution.RUN_ID_KEY, storeProviderId.queryRunId.toString)
Review Comment:
Yea prob dont need to set multiple times
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51955] Adding release() to ReadStateStore interface and reusing ReadStore for Streaming Aggregations [spark]
Posted by "ericm-db (via GitHub)" <gi...@apache.org>.
ericm-db commented on code in PR #50742:
URL: https://github.com/apache/spark/pull/50742#discussion_r2080067114
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala:
##########
@@ -365,6 +366,18 @@ private[sql] class RocksDBStateStoreProvider
}
result
}
+
+ override def release(): Unit = {
+ if (state != RELEASED) {
+ logInfo(log"Releasing ${MDC(VERSION_NUM, version + 1)} " +
Review Comment:
Hm, not sure but we log a similar line for abort [here](https://github.com/apache/spark/pull/50742/files#diff-76a55d7a6f0b91a19366e01c919c615a4d13afc1c1c37b017afa9df1c1705218L251)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51955] Adding release() to ReadStateStore interface and reusing ReadStore for Streaming Aggregations [spark]
Posted by "micheal-o (via GitHub)" <gi...@apache.org>.
micheal-o commented on code in PR #50742:
URL: https://github.com/apache/spark/pull/50742#discussion_r2078987001
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/HDFSBackedStateStoreProvider.scala:
##########
@@ -89,6 +89,8 @@ private[sql] class HDFSBackedStateStoreProvider extends StateStoreProvider with
override def abort(): Unit = {}
Review Comment:
ok, I think I remember a conversation we had about leaving it for now. But with this new change, can you check if there is still any non-test usage of calling abort on readstore?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51955] Adding release() to ReadStateStore interface and reusing ReadStore for Streaming Aggregations [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on code in PR #50742:
URL: https://github.com/apache/spark/pull/50742#discussion_r2067623925
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateStoreRDD.scala:
##########
@@ -82,19 +107,47 @@ class ReadStateStoreRDD[T: ClassTag, U: ClassTag](
useColumnFamilies: Boolean = false,
extraOptions: Map[String, String] = Map.empty)
extends BaseStateStoreRDD[T, U](dataRDD, checkpointLocation, queryRunId, operatorId,
- sessionState, storeCoordinator, extraOptions) {
+ sessionState, storeCoordinator, extraOptions) with StateStoreRDDProvider {
+
+ // Using a ConcurrentHashMap to track state stores by partition ID
+ // and whether this store was used to create a write store or not.
+ @transient private lazy val partitionStores =
+ new java.util.concurrent.ConcurrentHashMap[StateStoreProviderId, (ReadStateStore, Boolean)]()
+
+ override def getStateStoreForPartition(
+ providerId: StateStoreProviderId): Option[ReadStateStore] = {
+ val (readStore, _) = partitionStores.get(providerId)
+ partitionStores.put(providerId, (readStore, true))
+ Option(readStore)
+ }
override protected def getPartitions: Array[Partition] = dataRDD.partitions
override def compute(partition: Partition, ctxt: TaskContext): Iterator[U] = {
val storeProviderId = getStateProviderId(partition)
+ val partitionId = partition.index
val inputIter = dataRDD.iterator(partition, ctxt)
val store = StateStore.getReadOnly(
storeProviderId, keySchema, valueSchema, keyStateEncoderSpec, storeVersion,
- stateStoreCkptIds.map(_.apply(partition.index).head),
+ stateStoreCkptIds.map(_.apply(partitionId).head),
Review Comment:
Was this a bug before ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51955] Adding release() to ReadStateStore interface and reusing ReadStore for Streaming Aggregations [spark]
Posted by "liviazhu-db (via GitHub)" <gi...@apache.org>.
liviazhu-db commented on code in PR #50742:
URL: https://github.com/apache/spark/pull/50742#discussion_r2071995468
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/HDFSBackedStateStoreProvider.scala:
##########
@@ -194,6 +196,8 @@ private[sql] class HDFSBackedStateStoreProvider extends StateStoreProvider with
log"for ${MDC(LogKeys.STATE_STORE_PROVIDER, this)}")
}
+ override def release(): Unit = {}
Review Comment:
Can you update the state here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51955] Adding release() to ReadStateStore interface and reusing ReadStore for Streaming Aggregations [spark]
Posted by "ericm-db (via GitHub)" <gi...@apache.org>.
ericm-db closed pull request #50742: [SPARK-51955] Adding release() to ReadStateStore interface and reusing ReadStore for Streaming Aggregations
URL: https://github.com/apache/spark/pull/50742
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51955] Adding release() to ReadStateStore interface and reusing ReadStore for Streaming Aggregations [spark]
Posted by "liviazhu-db (via GitHub)" <gi...@apache.org>.
liviazhu-db commented on code in PR #50742:
URL: https://github.com/apache/spark/pull/50742#discussion_r2070612638
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala:
##########
@@ -446,17 +459,48 @@ private[sql] class RocksDBStateStoreProvider
override def stateStoreId: StateStoreId = stateStoreId_
- override def getStore(version: Long, uniqueId: Option[String] = None): StateStore = {
+ /**
+ * Creates and returns a state store with the specified parameters.
+ *
+ * @param version The version of the state store to load
+ * @param uniqueId Optional unique identifier for checkpoint
+ * @param readOnly Whether to open the store in read-only mode
+ * @param existingStore Optional existing store to reuse instead of creating a new one
+ * @return The loaded state store
+ */
+ private def loadStateStore(
+ version: Long,
+ uniqueId: Option[String],
+ readOnly: Boolean,
+ existingStore: Option[ReadStateStore] = None): StateStore = {
try {
if (version < 0) {
throw QueryExecutionErrors.unexpectedStateStoreVersion(version)
}
- rocksDB.load(
- version,
- stateStoreCkptId = if (storeConf.enableStateStoreCheckpointIds) uniqueId else None)
- new RocksDBStateStore(version)
- }
- catch {
+ try {
+ // Load RocksDB store
+ rocksDB.load(
+ version,
+ stateStoreCkptId = if (storeConf.enableStateStoreCheckpointIds) uniqueId else None,
+ readOnly = readOnly)
+
+ // Return appropriate store instance
+ existingStore match {
+ case Some(stateStore: RocksDBStateStore) =>
+ // Reuse existing store for getWriteStore case
+ StateStoreThreadLocalTracker.setUsedForWriteStore(true)
Review Comment:
Can you check if the returned object is the same as the passed in object in StateStoreRDD?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala:
##########
@@ -446,17 +459,48 @@ private[sql] class RocksDBStateStoreProvider
override def stateStoreId: StateStoreId = stateStoreId_
- override def getStore(version: Long, uniqueId: Option[String] = None): StateStore = {
+ /**
+ * Creates and returns a state store with the specified parameters.
+ *
+ * @param version The version of the state store to load
+ * @param uniqueId Optional unique identifier for checkpoint
+ * @param readOnly Whether to open the store in read-only mode
+ * @param existingStore Optional existing store to reuse instead of creating a new one
+ * @return The loaded state store
+ */
+ private def loadStateStore(
+ version: Long,
+ uniqueId: Option[String],
+ readOnly: Boolean,
+ existingStore: Option[ReadStateStore] = None): StateStore = {
try {
if (version < 0) {
throw QueryExecutionErrors.unexpectedStateStoreVersion(version)
}
- rocksDB.load(
- version,
- stateStoreCkptId = if (storeConf.enableStateStoreCheckpointIds) uniqueId else None)
- new RocksDBStateStore(version)
- }
- catch {
+ try {
+ // Load RocksDB store
+ rocksDB.load(
+ version,
+ stateStoreCkptId = if (storeConf.enableStateStoreCheckpointIds) uniqueId else None,
+ readOnly = readOnly)
+
+ // Return appropriate store instance
+ existingStore match {
+ case Some(stateStore: RocksDBStateStore) =>
+ // Reuse existing store for getWriteStore case
+ StateStoreThreadLocalTracker.setUsedForWriteStore(true)
Review Comment:
Can you check if the returned object is the same as the passed in object in StateStoreRDD and then set this if so?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51955] Adding release() to ReadStateStore interface and reusing ReadStore for Streaming Aggregations [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on code in PR #50742:
URL: https://github.com/apache/spark/pull/50742#discussion_r2070673830
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateStoreRDD.scala:
##########
@@ -27,6 +27,28 @@ import org.apache.spark.sql.internal.SessionState
import org.apache.spark.sql.types.StructType
import org.apache.spark.util.SerializableConfiguration
+/**
+ * Thread local storage for sharing StateStore instances between RDDs.
+ * This allows a ReadStateStore to be reused by a subsequent StateStore operation.
+ */
+object StateStoreThreadLocalTracker {
+ private val readStore: ThreadLocal[ReadStateStore] = new ThreadLocal[ReadStateStore]
Review Comment:
Can we combine these into a single thread local ? Just make them members of a case class ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51955] Adding release() to ReadStateStore interface and reusing ReadStore for Streaming Aggregations [spark]
Posted by "ericm-db (via GitHub)" <gi...@apache.org>.
ericm-db commented on PR #50742:
URL: https://github.com/apache/spark/pull/50742#issuecomment-2840486427
@anishshri-db @liviazhu-db Sure yeah sounds good
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51955] Adding release() to ReadStateStore interface and reusing ReadStore for Streaming Aggregations [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on code in PR #50742:
URL: https://github.com/apache/spark/pull/50742#discussion_r2067627737
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateStoreRDD.scala:
##########
@@ -82,19 +106,47 @@ class ReadStateStoreRDD[T: ClassTag, U: ClassTag](
useColumnFamilies: Boolean = false,
extraOptions: Map[String, String] = Map.empty)
extends BaseStateStoreRDD[T, U](dataRDD, checkpointLocation, queryRunId, operatorId,
- sessionState, storeCoordinator, extraOptions) {
+ sessionState, storeCoordinator, extraOptions) with StateStoreRDDProvider {
+
+ // Using a ConcurrentHashMap to track state stores by partition ID
+ // and whether this store was used to create a write store or not.
+ @transient private lazy val partitionStores =
+ new java.util.concurrent.ConcurrentHashMap[Int, (ReadStateStore, Boolean)]()
+
+ override def getStateStoreForPartition(partitionId: Int): Option[ReadStateStore] = {
+ val (readStore, _) = partitionStores.get(partitionId)
+ partitionStores.put(partitionId, (readStore, true))
+ Option(readStore)
+ }
override protected def getPartitions: Array[Partition] = dataRDD.partitions
override def compute(partition: Partition, ctxt: TaskContext): Iterator[U] = {
val storeProviderId = getStateProviderId(partition)
+ val partitionId = partition.index
val inputIter = dataRDD.iterator(partition, ctxt)
val store = StateStore.getReadOnly(
storeProviderId, keySchema, valueSchema, keyStateEncoderSpec, storeVersion,
- stateStoreCkptIds.map(_.apply(partition.index).head),
+ stateStoreCkptIds.map(_.apply(partitionId).head),
stateSchemaBroadcast,
useColumnFamilies, storeConf, hadoopConfBroadcast.value.value)
+
+ // Store reference for this partition
+ partitionStores.put(partitionId, (store, false))
+
+ // Register a cleanup callback to be executed when the task completes
+ ctxt.addTaskCompletionListener[Unit](_ => {
Review Comment:
@ericm-db - i guess we did not register any listener before in the `ReadStoreRDD` path ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51955] Adding release() to ReadStateStore interface and reusing ReadStore for Streaming Aggregations [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on code in PR #50742:
URL: https://github.com/apache/spark/pull/50742#discussion_r2067631028
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/package.scala:
##########
@@ -109,9 +110,16 @@ package object state {
val cleanedF = dataRDD.sparkContext.clean(storeReadFn)
val wrappedF = (store: ReadStateStore, iter: Iterator[T]) => {
// Clean up the state store.
- TaskContext.get().addTaskCompletionListener[Unit](_ => {
- store.abort()
+ val taskContext = TaskContext.get()
+ taskContext.addTaskCompletionListener[Unit](_ => {
Review Comment:
Why do we need it again here ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51955] Adding release() to ReadStateStore interface and reusing ReadStore for Streaming Aggregations [spark]
Posted by "liviazhu-db (via GitHub)" <gi...@apache.org>.
liviazhu-db commented on code in PR #50742:
URL: https://github.com/apache/spark/pull/50742#discussion_r2067549146
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala:
##########
@@ -468,29 +511,20 @@ private[sql] class RocksDBStateStoreProvider
case e: Throwable => throw QueryExecutionErrors.cannotLoadStore(e)
}
}
+ override def getStore(version: Long, uniqueId: Option[String] = None): StateStore = {
+ loadStateStore(version, uniqueId, readOnly = false)
+ }
+
+ override def getWriteStore(
+ readStore: ReadStateStore,
+ version: Long,
+ uniqueId: Option[String] = None): StateStore = {
+ assert(version == readStore.version)
+ loadStateStore(version, uniqueId, readOnly = false, existingStore = Some(readStore))
Review Comment:
Can you also verify that ProviderID (opId, partId, name) is the same?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51955] Adding release() to ReadStateStore interface and reusing ReadStore for Streaming Aggregations [spark]
Posted by "liviazhu-db (via GitHub)" <gi...@apache.org>.
liviazhu-db commented on code in PR #50742:
URL: https://github.com/apache/spark/pull/50742#discussion_r2070593834
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala:
##########
@@ -446,17 +459,48 @@ private[sql] class RocksDBStateStoreProvider
override def stateStoreId: StateStoreId = stateStoreId_
- override def getStore(version: Long, uniqueId: Option[String] = None): StateStore = {
+ /**
+ * Creates and returns a state store with the specified parameters.
+ *
+ * @param version The version of the state store to load
+ * @param uniqueId Optional unique identifier for checkpoint
+ * @param readOnly Whether to open the store in read-only mode
+ * @param existingStore Optional existing store to reuse instead of creating a new one
+ * @return The loaded state store
+ */
+ private def loadStateStore(
+ version: Long,
+ uniqueId: Option[String],
+ readOnly: Boolean,
+ existingStore: Option[ReadStateStore] = None): StateStore = {
try {
if (version < 0) {
throw QueryExecutionErrors.unexpectedStateStoreVersion(version)
}
- rocksDB.load(
- version,
- stateStoreCkptId = if (storeConf.enableStateStoreCheckpointIds) uniqueId else None)
- new RocksDBStateStore(version)
- }
- catch {
+ try {
+ // Load RocksDB store
+ rocksDB.load(
+ version,
+ stateStoreCkptId = if (storeConf.enableStateStoreCheckpointIds) uniqueId else None,
+ readOnly = readOnly)
+
+ // Return appropriate store instance
+ existingStore match {
+ case Some(stateStore: RocksDBStateStore) =>
+ // Reuse existing store for getWriteStore case
+ StateStoreThreadLocalTracker.setUsedForWriteStore(true)
Review Comment:
Can we set this in StateStoreRDD instead? Seems weird that we would be getting the existing state store from the fn args but setting the thread local directly form here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51955] Adding release() to ReadStateStore interface and reusing ReadStore for Streaming Aggregations [spark]
Posted by "ericm-db (via GitHub)" <gi...@apache.org>.
ericm-db commented on PR #50742:
URL: https://github.com/apache/spark/pull/50742#issuecomment-2840548725
cc @cloud-fan
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51955] Adding release() to ReadStateStore interface and reusing ReadStore for Streaming Aggregations [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on code in PR #50742:
URL: https://github.com/apache/spark/pull/50742#discussion_r2067602550
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala:
##########
@@ -468,29 +511,20 @@ private[sql] class RocksDBStateStoreProvider
case e: Throwable => throw QueryExecutionErrors.cannotLoadStore(e)
}
}
+ override def getStore(version: Long, uniqueId: Option[String] = None): StateStore = {
+ loadStateStore(version, uniqueId, readOnly = false)
+ }
+
+ override def getWriteStore(
+ readStore: ReadStateStore,
+ version: Long,
+ uniqueId: Option[String] = None): StateStore = {
+ assert(version == readStore.version)
Review Comment:
+1
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51955] Adding release() to ReadStateStore interface and reusing ReadStore for Streaming Aggregations [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on PR #50742:
URL: https://github.com/apache/spark/pull/50742#issuecomment-2840465177
@ericm-db - in the PR description, please update it to say what the new usage paradigm will look like as well
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51955] Adding release() to ReadStateStore interface and reusing ReadStore for Streaming Aggregations [spark]
Posted by "ericm-db (via GitHub)" <gi...@apache.org>.
ericm-db commented on code in PR #50742:
URL: https://github.com/apache/spark/pull/50742#discussion_r2070607523
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala:
##########
@@ -446,17 +459,48 @@ private[sql] class RocksDBStateStoreProvider
override def stateStoreId: StateStoreId = stateStoreId_
- override def getStore(version: Long, uniqueId: Option[String] = None): StateStore = {
+ /**
+ * Creates and returns a state store with the specified parameters.
+ *
+ * @param version The version of the state store to load
+ * @param uniqueId Optional unique identifier for checkpoint
+ * @param readOnly Whether to open the store in read-only mode
+ * @param existingStore Optional existing store to reuse instead of creating a new one
+ * @return The loaded state store
+ */
+ private def loadStateStore(
+ version: Long,
+ uniqueId: Option[String],
+ readOnly: Boolean,
+ existingStore: Option[ReadStateStore] = None): StateStore = {
try {
if (version < 0) {
throw QueryExecutionErrors.unexpectedStateStoreVersion(version)
}
- rocksDB.load(
- version,
- stateStoreCkptId = if (storeConf.enableStateStoreCheckpointIds) uniqueId else None)
- new RocksDBStateStore(version)
- }
- catch {
+ try {
+ // Load RocksDB store
+ rocksDB.load(
+ version,
+ stateStoreCkptId = if (storeConf.enableStateStoreCheckpointIds) uniqueId else None,
+ readOnly = readOnly)
+
+ // Return appropriate store instance
+ existingStore match {
+ case Some(stateStore: RocksDBStateStore) =>
+ // Reuse existing store for getWriteStore case
+ StateStoreThreadLocalTracker.setUsedForWriteStore(true)
Review Comment:
Yeah I wanted to - but I think there would be some abstraction leak (type checking within StateStoreRDD) that I wanted to avoid, since we don't 'reuse' stores for HDFS, we just open a new one.
The reason for this is because HDFSBackedStateStore and HDFSBackedReadStateStore are different classes, so we can't directly return the store like we do [here](https://github.com/apache/spark/pull/50742/files#diff-76a55d7a6f0b91a19366e01c919c615a4d13afc1c1c37b017afa9df1c1705218R492)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51955] Adding release() to ReadStateStore interface and reusing ReadStore for Streaming Aggregations [spark]
Posted by "ericm-db (via GitHub)" <gi...@apache.org>.
ericm-db commented on code in PR #50742:
URL: https://github.com/apache/spark/pull/50742#discussion_r2070615297
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala:
##########
@@ -446,17 +459,48 @@ private[sql] class RocksDBStateStoreProvider
override def stateStoreId: StateStoreId = stateStoreId_
- override def getStore(version: Long, uniqueId: Option[String] = None): StateStore = {
+ /**
+ * Creates and returns a state store with the specified parameters.
+ *
+ * @param version The version of the state store to load
+ * @param uniqueId Optional unique identifier for checkpoint
+ * @param readOnly Whether to open the store in read-only mode
+ * @param existingStore Optional existing store to reuse instead of creating a new one
+ * @return The loaded state store
+ */
+ private def loadStateStore(
+ version: Long,
+ uniqueId: Option[String],
+ readOnly: Boolean,
+ existingStore: Option[ReadStateStore] = None): StateStore = {
try {
if (version < 0) {
throw QueryExecutionErrors.unexpectedStateStoreVersion(version)
}
- rocksDB.load(
- version,
- stateStoreCkptId = if (storeConf.enableStateStoreCheckpointIds) uniqueId else None)
- new RocksDBStateStore(version)
- }
- catch {
+ try {
+ // Load RocksDB store
+ rocksDB.load(
+ version,
+ stateStoreCkptId = if (storeConf.enableStateStoreCheckpointIds) uniqueId else None,
+ readOnly = readOnly)
+
+ // Return appropriate store instance
+ existingStore match {
+ case Some(stateStore: RocksDBStateStore) =>
+ // Reuse existing store for getWriteStore case
+ StateStoreThreadLocalTracker.setUsedForWriteStore(true)
Review Comment:
Ah that would be a good way to go about this - let me try
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51955] Adding release() to ReadStateStore interface and reusing ReadStore for Streaming Aggregations [spark]
Posted by "ericm-db (via GitHub)" <gi...@apache.org>.
ericm-db commented on PR #50742:
URL: https://github.com/apache/spark/pull/50742#issuecomment-2845440177
> Looks good! Could you add a test in StateStoreRDDSuite to check the ThreadLocal logic correctly passes the readstore to the writestore too?
Yup, working on that rn!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51955] Adding release() to ReadStateStore interface and reusing ReadStore for Streaming Aggregations [spark]
Posted by "liviazhu-db (via GitHub)" <gi...@apache.org>.
liviazhu-db commented on code in PR #50742:
URL: https://github.com/apache/spark/pull/50742#discussion_r2071995468
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/HDFSBackedStateStoreProvider.scala:
##########
@@ -194,6 +196,8 @@ private[sql] class HDFSBackedStateStoreProvider extends StateStoreProvider with
log"for ${MDC(LogKeys.STATE_STORE_PROVIDER, this)}")
}
+ override def release(): Unit = {}
Review Comment:
Can you add a new state and update the state here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51955] Adding release() to ReadStateStore interface and reusing ReadStore for Streaming Aggregations [spark]
Posted by "ericm-db (via GitHub)" <gi...@apache.org>.
ericm-db commented on code in PR #50742:
URL: https://github.com/apache/spark/pull/50742#discussion_r2067627565
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateStoreRDD.scala:
##########
@@ -82,19 +107,47 @@ class ReadStateStoreRDD[T: ClassTag, U: ClassTag](
useColumnFamilies: Boolean = false,
extraOptions: Map[String, String] = Map.empty)
extends BaseStateStoreRDD[T, U](dataRDD, checkpointLocation, queryRunId, operatorId,
- sessionState, storeCoordinator, extraOptions) {
+ sessionState, storeCoordinator, extraOptions) with StateStoreRDDProvider {
+
+ // Using a ConcurrentHashMap to track state stores by partition ID
+ // and whether this store was used to create a write store or not.
+ @transient private lazy val partitionStores =
+ new java.util.concurrent.ConcurrentHashMap[StateStoreProviderId, (ReadStateStore, Boolean)]()
+
+ override def getStateStoreForPartition(
+ providerId: StateStoreProviderId): Option[ReadStateStore] = {
+ val (readStore, _) = partitionStores.get(providerId)
+ partitionStores.put(providerId, (readStore, true))
+ Option(readStore)
+ }
override protected def getPartitions: Array[Partition] = dataRDD.partitions
override def compute(partition: Partition, ctxt: TaskContext): Iterator[U] = {
val storeProviderId = getStateProviderId(partition)
+ val partitionId = partition.index
val inputIter = dataRDD.iterator(partition, ctxt)
val store = StateStore.getReadOnly(
storeProviderId, keySchema, valueSchema, keyStateEncoderSpec, storeVersion,
- stateStoreCkptIds.map(_.apply(partition.index).head),
+ stateStoreCkptIds.map(_.apply(partitionId).head),
Review Comment:
No these are equivalent though. I'll change it back
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org