You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "anishshri-db (via GitHub)" <gi...@apache.org> on 2024/02/06 04:15:18 UTC
[PR] [SPARK-46979] Add support for specifying key and value encoder separately and also for each col family in RocksDB state store provider [spark]
anishshri-db opened a new pull request, #45038:
URL: https://github.com/apache/spark/pull/45038
### What changes were proposed in this pull request?
Add support for specifying key and value encoder separately and also for each col family in RocksDB state store provider
### Why are the changes needed?
This change allows us to specify encoder for key/values separately and avoid encoding additional bytes. Also, it allows us to set schemas/encoders for individual column families, which will be required for future changes related to transformWithState operator.
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
Added unit tests
```
[info] Run completed in 3 minutes, 5 seconds.
[info] Total number of tests run: 286
[info] Suites: completed 1, aborted 0
[info] Tests: succeeded 286, failed 0, canceled 0, ignored 0, pending 0
[info] All tests passed.
```
### Was this patch authored or co-authored using generative AI tooling?
No
--
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-46979][SS] Add support for specifying key and value encoder separately and also for each col family in RocksDB state store provider [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #45038:
URL: https://github.com/apache/spark/pull/45038#discussion_r1487154735
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala:
##########
@@ -92,10 +108,17 @@ private[sql] class RocksDBStateStoreProvider
override def prefixScan(prefixKey: UnsafeRow, colFamilyName: String):
Iterator[UnsafeRowPair] = {
- require(encoder.supportPrefixKeyScan, "Prefix scan requires setting prefix key!")
-
- val prefix = encoder.encodePrefixKey(prefixKey)
- rocksDB.prefixScan(prefix, colFamilyName).map(kv => encoder.decode(kv))
+ val kvEncoder = keyValueEncoderMap.get(colFamilyName)
+ require(kvEncoder._1.supportPrefixKeyScan,
+ "Prefix scan requires setting prefix key!")
+
+ val prefix = kvEncoder._1.encodePrefixKey(prefixKey)
+ rocksDB.prefixScan(prefix, colFamilyName).map { kv =>
+ val rowPair = new UnsafeRowPair()
Review Comment:
Please create this outside of lambda if it works - if we are creating pair object per kv, it's no meaning of reusing this.
--
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-46979][SS] Add support for specifying key and value encoder separately and also for each col family in RocksDB state store provider [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #45038:
URL: https://github.com/apache/spark/pull/45038#discussion_r1487155443
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala:
##########
@@ -69,19 +79,25 @@ private[sql] class RocksDBStateStoreProvider
verify(state == UPDATING, "Cannot put after already committed or aborted")
verify(key != null, "Key cannot be null")
require(value != null, "Cannot put a null value")
- rocksDB.put(encoder.encodeKey(key), encoder.encodeValue(value), colFamilyName)
+ val kvEncoder = keyValueEncoderMap.get(colFamilyName)
+ rocksDB.put(kvEncoder._1.encodeKey(key),
+ kvEncoder._2.encodeValue(value), colFamilyName)
}
override def remove(key: UnsafeRow, colFamilyName: String): Unit = {
verify(state == UPDATING, "Cannot remove after already committed or aborted")
verify(key != null, "Key cannot be null")
- rocksDB.remove(encoder.encodeKey(key), colFamilyName)
+ val kvEncoder = keyValueEncoderMap.get(colFamilyName)
+ rocksDB.remove(kvEncoder._1.encodeKey(key), colFamilyName)
}
override def iterator(colFamilyName: String): Iterator[UnsafeRowPair] = {
+ val kvEncoder = keyValueEncoderMap.get(colFamilyName)
rocksDB.iterator(colFamilyName).map { kv =>
- val rowPair = encoder.decode(kv)
- if (!isValidated && rowPair.value != null) {
+ val rowPair = new UnsafeRowPair()
Review Comment:
same here (sorry I reviewed backward)
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala:
##########
@@ -215,7 +240,9 @@ private[sql] class RocksDBStateStoreProvider
(keySchema.length > numColsPrefixKey), "The number of columns in the key must be " +
"greater than the number of columns for prefix key!")
- this.encoder = RocksDBStateEncoder.getEncoder(keySchema, valueSchema, numColsPrefixKey)
+ keyValueEncoderMap.putIfAbsent(StateStore.DEFAULT_COL_FAMILY_NAME,
Review Comment:
(Maybe microbenchmark could tell that this could regress for default column family only - map lookup with carefully crafted lock operation in every op, though I'd rather not concern before we see actual regression.)
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala:
##########
@@ -287,7 +314,8 @@ private[sql] class RocksDBStateStoreProvider
useColumnFamilies)
}
- @volatile private var encoder: RocksDBStateEncoder = _
+ @volatile private var keyValueEncoderMap = new java.util.concurrent.ConcurrentHashMap[String,
Review Comment:
nit: could be `val`, and also probably no need to be `@volatile`.
--
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-46979][SS] Add support for specifying key and value encoder separately and also for each col family in RocksDB state store provider [spark]
Posted by "sahnib (via GitHub)" <gi...@apache.org>.
sahnib commented on code in PR #45038:
URL: https://github.com/apache/spark/pull/45038#discussion_r1480416349
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala:
##########
@@ -287,7 +332,13 @@ private[sql] class RocksDBStateStoreProvider
useColumnFamilies)
}
- @volatile private var encoder: RocksDBStateEncoder = _
+ private val encoderMapLock = new Object
Review Comment:
[nit] rename this to `columnFamilyEncoderMapLock` as it only applies to column families. Encoders for default column family are picked at State store init.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala:
##########
@@ -287,7 +332,13 @@ private[sql] class RocksDBStateStoreProvider
useColumnFamilies)
}
- @volatile private var encoder: RocksDBStateEncoder = _
+ private val encoderMapLock = new Object
+
+ @GuardedBy("encoderMapLock")
+ @volatile private var keyEncoderMap = new mutable.HashMap[String, RocksDBKeyStateEncoder]
+
+ @GuardedBy("encoderMapLock")
+ @volatile private var valueEncoderMap = new mutable.HashMap[String, RocksDBValueStateEncoder]
Review Comment:
Can we just use a concurrentHashMap with signature `Map[String, (RocksDBKeyStateEncoder, RocksDBValueStateEncoder)]`. Then, we don't need to synchronize using a lock.
--
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-46979][SS] Add support for specifying key and value encoder separately and also for each col family in RocksDB state store provider [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on code in PR #45038:
URL: https://github.com/apache/spark/pull/45038#discussion_r1487252966
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala:
##########
@@ -215,7 +240,9 @@ private[sql] class RocksDBStateStoreProvider
(keySchema.length > numColsPrefixKey), "The number of columns in the key must be " +
"greater than the number of columns for prefix key!")
- this.encoder = RocksDBStateEncoder.getEncoder(keySchema, valueSchema, numColsPrefixKey)
+ keyValueEncoderMap.putIfAbsent(StateStore.DEFAULT_COL_FAMILY_NAME,
Review Comment:
Ah ok - yea mainly didn't want to maintain 2 data structures for this. But if we find that its more expensive, then we can just split some of the logic for the default col family case
--
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-46979][SS] Add support for specifying key and value encoder separately and also for each col family in RocksDB state store provider [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on code in PR #45038:
URL: https://github.com/apache/spark/pull/45038#discussion_r1487255889
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/ValueStateImpl.scala:
##########
@@ -54,18 +56,15 @@ class ValueStateImpl[S](
throw StateStoreErrors.implicitKeyNotFound(stateName)
}
- val toRow = keyExprEnc.createSerializer()
- val keyByteArr = toRow
+ val keyByteArr = keyExprEnc.createSerializer()
Review Comment:
Done
--
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-46979][SS] Add support for specifying key and value encoder separately and also for each col family in RocksDB state store provider [spark]
Posted by "sahnib (via GitHub)" <gi...@apache.org>.
sahnib commented on code in PR #45038:
URL: https://github.com/apache/spark/pull/45038#discussion_r1487141005
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala:
##########
@@ -48,54 +50,86 @@ private[sql] class RocksDBStateStoreProvider
override def version: Long = lastVersion
- override def createColFamilyIfAbsent(colFamilyName: String): Unit = {
+ override def createColFamilyIfAbsent(
+ colFamilyName: String,
+ keySchema: StructType,
+ numColsPrefixKey: Int,
+ valueSchema: StructType): Unit = {
verify(colFamilyName != StateStore.DEFAULT_COL_FAMILY_NAME,
s"Failed to create column family with reserved_name=$colFamilyName")
+ verify(useColumnFamilies, "Column families are not supported in this store")
rocksDB.createColFamilyIfAbsent(colFamilyName)
+ encoderMapLock.synchronized {
+ keyEncoderMap.getOrElseUpdate(colFamilyName,
+ RocksDBStateEncoder.getKeyEncoder(keySchema, numColsPrefixKey))
+
+ valueEncoderMap.getOrElseUpdate(colFamilyName,
+ RocksDBStateEncoder.getValueEncoder(valueSchema))
+ }
Review Comment:
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-46979][SS] Add support for specifying key and value encoder separately and also for each col family in RocksDB state store provider [spark]
Posted by "sahnib (via GitHub)" <gi...@apache.org>.
sahnib commented on code in PR #45038:
URL: https://github.com/apache/spark/pull/45038#discussion_r1480430719
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala:
##########
@@ -48,54 +50,86 @@ private[sql] class RocksDBStateStoreProvider
override def version: Long = lastVersion
- override def createColFamilyIfAbsent(colFamilyName: String): Unit = {
+ override def createColFamilyIfAbsent(
+ colFamilyName: String,
+ keySchema: StructType,
+ numColsPrefixKey: Int,
+ valueSchema: StructType): Unit = {
verify(colFamilyName != StateStore.DEFAULT_COL_FAMILY_NAME,
s"Failed to create column family with reserved_name=$colFamilyName")
+ verify(useColumnFamilies, "Column families are not supported in this store")
rocksDB.createColFamilyIfAbsent(colFamilyName)
+ encoderMapLock.synchronized {
+ keyEncoderMap.getOrElseUpdate(colFamilyName,
+ RocksDBStateEncoder.getKeyEncoder(keySchema, numColsPrefixKey))
+
+ valueEncoderMap.getOrElseUpdate(colFamilyName,
+ RocksDBStateEncoder.getValueEncoder(valueSchema))
+ }
Review Comment:
Should we throw an exception if the passed key/value schema are now different for a existing column family (which was previously created).
Consider the scenario below:
1. User creates a colFamily with keySchema K, valueSchema V.
2. User issues the call again with keySchema K1, valueSchema V1. Note K1 != K, or V1 != V.
3. Now the call in (2) succeeds, but encoders are using a different schema.
--
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-46979][SS] Add support for specifying key and value encoder separately and also for each col family in RocksDB state store provider [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #45038:
URL: https://github.com/apache/spark/pull/45038#discussion_r1487224566
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/ValueStateImpl.scala:
##########
@@ -52,14 +58,12 @@ class ValueStateImpl[S](
val keyByteArr = toRow
.apply(keyOption.get).asInstanceOf[UnsafeRow].getBytes()
- val schemaForKeyRow: StructType = new StructType().add("key", BinaryType)
val keyEncoder = UnsafeProjection.create(schemaForKeyRow)
Review Comment:
OK to do in separate PR, as it's technically not in this scope of PR.
--
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-46979][SS] Add support for specifying key and value encoder separately and also for each col family in RocksDB state store provider [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on code in PR #45038:
URL: https://github.com/apache/spark/pull/45038#discussion_r1487240502
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/ValueStateImpl.scala:
##########
@@ -52,14 +58,12 @@ class ValueStateImpl[S](
val keyByteArr = toRow
.apply(keyOption.get).asInstanceOf[UnsafeRow].getBytes()
- val schemaForKeyRow: StructType = new StructType().add("key", BinaryType)
val keyEncoder = UnsafeProjection.create(schemaForKeyRow)
Review Comment:
Done
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala:
##########
@@ -92,10 +108,17 @@ private[sql] class RocksDBStateStoreProvider
override def prefixScan(prefixKey: UnsafeRow, colFamilyName: String):
Iterator[UnsafeRowPair] = {
- require(encoder.supportPrefixKeyScan, "Prefix scan requires setting prefix key!")
-
- val prefix = encoder.encodePrefixKey(prefixKey)
- rocksDB.prefixScan(prefix, colFamilyName).map(kv => encoder.decode(kv))
+ val kvEncoder = keyValueEncoderMap.get(colFamilyName)
+ require(kvEncoder._1.supportPrefixKeyScan,
+ "Prefix scan requires setting prefix key!")
+
+ val prefix = kvEncoder._1.encodePrefixKey(prefixKey)
+ rocksDB.prefixScan(prefix, colFamilyName).map { kv =>
+ val rowPair = new UnsafeRowPair()
Review Comment:
Done
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala:
##########
@@ -69,19 +79,25 @@ private[sql] class RocksDBStateStoreProvider
verify(state == UPDATING, "Cannot put after already committed or aborted")
verify(key != null, "Key cannot be null")
require(value != null, "Cannot put a null value")
- rocksDB.put(encoder.encodeKey(key), encoder.encodeValue(value), colFamilyName)
+ val kvEncoder = keyValueEncoderMap.get(colFamilyName)
+ rocksDB.put(kvEncoder._1.encodeKey(key),
+ kvEncoder._2.encodeValue(value), colFamilyName)
}
override def remove(key: UnsafeRow, colFamilyName: String): Unit = {
verify(state == UPDATING, "Cannot remove after already committed or aborted")
verify(key != null, "Key cannot be null")
- rocksDB.remove(encoder.encodeKey(key), colFamilyName)
+ val kvEncoder = keyValueEncoderMap.get(colFamilyName)
+ rocksDB.remove(kvEncoder._1.encodeKey(key), colFamilyName)
}
override def iterator(colFamilyName: String): Iterator[UnsafeRowPair] = {
+ val kvEncoder = keyValueEncoderMap.get(colFamilyName)
rocksDB.iterator(colFamilyName).map { kv =>
- val rowPair = encoder.decode(kv)
- if (!isValidated && rowPair.value != null) {
+ val rowPair = new UnsafeRowPair()
Review Comment:
Done
--
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-46979][SS] Add support for specifying key and value encoder separately and also for each col family in RocksDB state store provider [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on code in PR #45038:
URL: https://github.com/apache/spark/pull/45038#discussion_r1487242677
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala:
##########
@@ -215,7 +240,9 @@ private[sql] class RocksDBStateStoreProvider
(keySchema.length > numColsPrefixKey), "The number of columns in the key must be " +
"greater than the number of columns for prefix key!")
- this.encoder = RocksDBStateEncoder.getEncoder(keySchema, valueSchema, numColsPrefixKey)
+ keyValueEncoderMap.putIfAbsent(StateStore.DEFAULT_COL_FAMILY_NAME,
Review Comment:
Yea I didn't worry about it too much, given that the provider init likely happens once for long lived queries and where we can retain the use of the same provider on the same executor across m/batch executions.
--
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-46979][SS] Add support for specifying key and value encoder separately and also for each col family in RocksDB state store provider [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #45038:
URL: https://github.com/apache/spark/pull/45038#discussion_r1487246611
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala:
##########
@@ -215,7 +240,9 @@ private[sql] class RocksDBStateStoreProvider
(keySchema.length > numColsPrefixKey), "The number of columns in the key must be " +
"greater than the number of columns for prefix key!")
- this.encoder = RocksDBStateEncoder.getEncoder(keySchema, valueSchema, numColsPrefixKey)
+ keyValueEncoderMap.putIfAbsent(StateStore.DEFAULT_COL_FAMILY_NAME,
Review Comment:
No, what I meant is to look up concurrent map per "every op" to figure out encoder, for existing stateful operators - previously it was just a reference to the field. But ops is relatively very cheap compared to commit as of now, so let's see.
--
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-46979][SS] Add support for specifying key and value encoder separately and also for each col family in RocksDB state store provider [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on PR #45038:
URL: https://github.com/apache/spark/pull/45038#issuecomment-1942570003
Thanks! Merging to master.
--
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-46979][SS] Add support for specifying key and value encoder separately and also for each col family in RocksDB state store provider [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #45038:
URL: https://github.com/apache/spark/pull/45038#discussion_r1487246611
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala:
##########
@@ -215,7 +240,9 @@ private[sql] class RocksDBStateStoreProvider
(keySchema.length > numColsPrefixKey), "The number of columns in the key must be " +
"greater than the number of columns for prefix key!")
- this.encoder = RocksDBStateEncoder.getEncoder(keySchema, valueSchema, numColsPrefixKey)
+ keyValueEncoderMap.putIfAbsent(StateStore.DEFAULT_COL_FAMILY_NAME,
Review Comment:
No, what I meant is to look up concurrent map per "every op" to figure out encoder - previously it was just a reference to the field. But ops is relatively very cheap compared to commit as of now, so let's see.
--
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-46979][SS] Add support for specifying key and value encoder separately and also for each col family in RocksDB state store provider [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #45038:
URL: https://github.com/apache/spark/pull/45038#discussion_r1487249005
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/ValueStateImpl.scala:
##########
@@ -54,18 +56,15 @@ class ValueStateImpl[S](
throw StateStoreErrors.implicitKeyNotFound(stateName)
}
- val toRow = keyExprEnc.createSerializer()
- val keyByteArr = toRow
+ val keyByteArr = keyExprEnc.createSerializer()
Review Comment:
nit: sorry for missing in initial review - this seems to be also something we can move out
--
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-46979][SS] Add support for specifying key and value encoder separately and also for each col family in RocksDB state store provider [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on code in PR #45038:
URL: https://github.com/apache/spark/pull/45038#discussion_r1480570588
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala:
##########
@@ -48,54 +50,86 @@ private[sql] class RocksDBStateStoreProvider
override def version: Long = lastVersion
- override def createColFamilyIfAbsent(colFamilyName: String): Unit = {
+ override def createColFamilyIfAbsent(
+ colFamilyName: String,
+ keySchema: StructType,
+ numColsPrefixKey: Int,
+ valueSchema: StructType): Unit = {
verify(colFamilyName != StateStore.DEFAULT_COL_FAMILY_NAME,
s"Failed to create column family with reserved_name=$colFamilyName")
+ verify(useColumnFamilies, "Column families are not supported in this store")
rocksDB.createColFamilyIfAbsent(colFamilyName)
+ encoderMapLock.synchronized {
+ keyEncoderMap.getOrElseUpdate(colFamilyName,
+ RocksDBStateEncoder.getKeyEncoder(keySchema, numColsPrefixKey))
+
+ valueEncoderMap.getOrElseUpdate(colFamilyName,
+ RocksDBStateEncoder.getValueEncoder(valueSchema))
+ }
Review Comment:
If the entry exists, then we won't replace it actually with the current changes. But on restart, this can happen. However, for the col family case - we need to decide interaction with state schema compatibility checker anyway. So will prefer to do this in a separate PR
--
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-46979][SS] Add support for specifying key and value encoder separately and also for each col family in RocksDB state store provider [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on code in PR #45038:
URL: https://github.com/apache/spark/pull/45038#discussion_r1487240784
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala:
##########
@@ -287,7 +314,8 @@ private[sql] class RocksDBStateStoreProvider
useColumnFamilies)
}
- @volatile private var encoder: RocksDBStateEncoder = _
+ @volatile private var keyValueEncoderMap = new java.util.concurrent.ConcurrentHashMap[String,
Review Comment:
Done
--
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-46979][SS] Add support for specifying key and value encoder separately and also for each col family in RocksDB state store provider [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #45038:
URL: https://github.com/apache/spark/pull/45038#discussion_r1487249005
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/ValueStateImpl.scala:
##########
@@ -54,18 +56,15 @@ class ValueStateImpl[S](
throw StateStoreErrors.implicitKeyNotFound(stateName)
}
- val toRow = keyExprEnc.createSerializer()
- val keyByteArr = toRow
+ val keyByteArr = keyExprEnc.createSerializer()
Review Comment:
nit: sorry for missing - this seems to be also something we can move out
--
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-46979][SS] Add support for specifying key and value encoder separately and also for each col family in RocksDB state store provider [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #45038:
URL: https://github.com/apache/spark/pull/45038#discussion_r1487149178
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/ValueStateImpl.scala:
##########
@@ -52,14 +58,12 @@ class ValueStateImpl[S](
val keyByteArr = toRow
.apply(keyOption.get).asInstanceOf[UnsafeRow].getBytes()
- val schemaForKeyRow: StructType = new StructType().add("key", BinaryType)
val keyEncoder = UnsafeProjection.create(schemaForKeyRow)
Review Comment:
This seems to be missing spot on previous PR review - we need to do this only once instead of each method call. Please move to the field of class. Same for valueEncoder for encodeValue 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-46979][SS] Add support for specifying key and value encoder separately and also for each col family in RocksDB state store provider [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR closed pull request #45038: [SPARK-46979][SS] Add support for specifying key and value encoder separately and also for each col family in RocksDB state store provider
URL: https://github.com/apache/spark/pull/45038
--
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-46979][SS] Add support for specifying key and value encoder separately and also for each col family in RocksDB state store provider [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on code in PR #45038:
URL: https://github.com/apache/spark/pull/45038#discussion_r1480427117
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala:
##########
@@ -287,7 +332,13 @@ private[sql] class RocksDBStateStoreProvider
useColumnFamilies)
}
- @volatile private var encoder: RocksDBStateEncoder = _
+ private val encoderMapLock = new Object
Review Comment:
Not sure I understand - we would still have to protect the same hashMap right ?
--
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-46979][SS] Add support for specifying key and value encoder separately and also for each col family in RocksDB state store provider [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on code in PR #45038:
URL: https://github.com/apache/spark/pull/45038#discussion_r1480567343
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala:
##########
@@ -287,7 +332,13 @@ private[sql] class RocksDBStateStoreProvider
useColumnFamilies)
}
- @volatile private var encoder: RocksDBStateEncoder = _
+ private val encoderMapLock = new Object
+
+ @GuardedBy("encoderMapLock")
+ @volatile private var keyEncoderMap = new mutable.HashMap[String, RocksDBKeyStateEncoder]
+
+ @GuardedBy("encoderMapLock")
+ @volatile private var valueEncoderMap = new mutable.HashMap[String, RocksDBValueStateEncoder]
Review Comment:
Done
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala:
##########
@@ -287,7 +332,13 @@ private[sql] class RocksDBStateStoreProvider
useColumnFamilies)
}
- @volatile private var encoder: RocksDBStateEncoder = _
+ private val encoderMapLock = new Object
Review Comment:
Removed this
--
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-46979][SS] Add support for specifying key and value encoder separately and also for each col family in RocksDB state store provider [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on PR #45038:
URL: https://github.com/apache/spark/pull/45038#issuecomment-1928757411
@HeartSaVioR - PTAL, 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