This is an automated email from the ASF dual-hosted git repository.
HeartSaVioR pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 3aba75525ff3 [SPARK-58589][SS] Drop redundant .apply on Function1
encoders in StateTypesEncoderUtils
3aba75525ff3 is described below
commit 3aba75525ff3aa65cac0c5897af6794f719e3372
Author: Uros Bojanic <[email protected]>
AuthorDate: Thu Aug 6 15:42:15 2026 +0900
[SPARK-58589][SS] Drop redundant .apply on Function1 encoders in
StateTypesEncoderUtils
### What changes were proposed in this pull request?
Replaces the explicit `.apply(...)` calls with direct invocation on the
encoder serializers, deserializers and unsafe projections in
`StateTypesEncoderUtils` (19 call sites).
### Why are the changes needed?
All of these receivers are `Function1` subtypes:
`ExpressionEncoder.Serializer` extends `(T => InternalRow)`,
`ExpressionEncoder.Deserializer` extends `(InternalRow => T)`, and
`UnsafeProjection` extends `Projection`, which extends `(InternalRow =>
InternalRow)`. For a `Function1`, `f(x)` is exactly `f.apply(x)`, so the
explicit form adds noise without meaning. The same file already calls several
projections directly, so this removes an internal inconsistency rather than
introducing a [...]
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Syntactic change with identical semantics; existing transformWithState
tests cover these paths. No new tests needed.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8)
Closes #57792 from uros-b/style-statetypesencoder-apply.
Authored-by: Uros Bojanic <[email protected]>
Signed-off-by: Jungtaek Lim <[email protected]>
(cherry picked from commit 41144ce2c5076698241ac04bc433d803e842d608)
Signed-off-by: Jungtaek Lim <[email protected]>
---
.../StateTypesEncoderUtils.scala | 38 +++++++++++-----------
1 file changed, 19 insertions(+), 19 deletions(-)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/operators/stateful/transformwithstate/StateTypesEncoderUtils.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/operators/stateful/transformwithstate/StateTypesEncoderUtils.scala
index d147ad66c246..480aebfb351f 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/operators/stateful/transformwithstate/StateTypesEncoderUtils.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/operators/stateful/transformwithstate/StateTypesEncoderUtils.scala
@@ -117,14 +117,14 @@ class StateTypesEncoder[V](
throw StateStoreErrors.implicitKeyNotFound(stateName)
}
- keySerializer.apply(keyOption.get).asInstanceOf[UnsafeRow]
+ keySerializer(keyOption.get).asInstanceOf[UnsafeRow]
}
/**
* Encode the specified value in Spark UnsafeRow with no ttl.
*/
def encodeValue(value: V): UnsafeRow = {
- objToRowSerializer.apply(value).asInstanceOf[UnsafeRow]
+ objToRowSerializer(value).asInstanceOf[UnsafeRow]
}
/**
@@ -132,15 +132,15 @@ class StateTypesEncoder[V](
* with provided ttl expiration.
*/
def encodeValue(value: V, expirationMs: Long): UnsafeRow = {
- val objRow: InternalRow = objToRowSerializer.apply(value)
- valueTTLProjection.apply(InternalRow(objRow, expirationMs))
+ val objRow: InternalRow = objToRowSerializer(value)
+ valueTTLProjection(InternalRow(objRow, expirationMs))
}
def decodeValue(row: UnsafeRow): V = {
if (hasTtl) {
- rowToObjDeserializer.apply(row.getStruct(0, valEncoder.schema.length))
+ rowToObjDeserializer(row.getStruct(0, valEncoder.schema.length))
} else {
- rowToObjDeserializer.apply(row)
+ rowToObjDeserializer(row)
}
}
@@ -213,14 +213,14 @@ class CompositeKeyStateEncoder[K, V](
throw StateStoreErrors.implicitKeyNotFound(stateName)
}
val groupingKey = keyOption.get
- val groupingKeyRow = groupingKeySerializer.apply(groupingKey)
+ val groupingKeyRow = groupingKeySerializer(groupingKey)
// Create the final unsafeRow mapping column name "key" to the keyRow
groupingKeyProjection(InternalRow(groupingKeyRow))
}
def encodeUserKey(userKey: K): UnsafeRow = {
- val userKeyRow = userKeySerializer.apply(userKey)
+ val userKeyRow = userKeySerializer(userKey)
// Create the final unsafeRow mapping column name "userKey" to the
userKeyRow
userKeyProjection(InternalRow(userKeyRow))
@@ -237,8 +237,8 @@ class CompositeKeyStateEncoder[K, V](
}
val groupingKey = keyOption.get
- val keyRow = groupingKeySerializer.apply(groupingKey)
- val userKeyRow = userKeySerializer.apply(userKey)
+ val keyRow = groupingKeySerializer(groupingKey)
+ val userKeyRow = userKeySerializer(userKey)
// Create the final unsafeRow combining the keyRow and userKeyRow
compositeKeyProjection(InternalRow(keyRow, userKeyRow))
@@ -253,7 +253,7 @@ class CompositeKeyStateEncoder[K, V](
* Only user key is returned though grouping key also exist in the row.
*/
def decodeCompositeKey(row: UnsafeRow): K = {
- userKeyRowToObjDeserializer.apply(row.getStruct(1,
userKeyEnc.schema.length))
+ userKeyRowToObjDeserializer(row.getStruct(1, userKeyEnc.schema.length))
}
}
@@ -264,7 +264,7 @@ class TTLEncoder(schema: StructType) {
// Take a groupingKey UnsafeRow and turn it into a (expirationMs,
groupingKey) UnsafeRow.
def encodeTTLRow(expirationMs: Long, elementKey: UnsafeRow): UnsafeRow = {
- ttlKeyProjection.apply(
+ ttlKeyProjection(
InternalRow(expirationMs, elementKey.asInstanceOf[InternalRow]))
}
}
@@ -294,22 +294,22 @@ class TimerKeyEncoder(keyExprEnc: ExpressionEncoder[Any])
{
private val secIndexKeyProjection =
UnsafeProjection.create(keySchemaForSecIndex)
def encodedKey(groupingKey: Any, expiryTimestampMs: Long): UnsafeRow = {
- val keyRow = keySerializer.apply(groupingKey)
- keyRowProjection.apply(InternalRow(keyRow, expiryTimestampMs))
+ val keyRow = keySerializer(groupingKey)
+ keyRowProjection(InternalRow(keyRow, expiryTimestampMs))
}
def encodeSecIndexKey(groupingKey: Any, expiryTimestampMs: Long): UnsafeRow
= {
- val keyRow = keySerializer.apply(groupingKey)
- secIndexKeyProjection.apply(InternalRow(expiryTimestampMs, keyRow))
+ val keyRow = keySerializer(groupingKey)
+ secIndexKeyProjection(InternalRow(expiryTimestampMs, keyRow))
}
def encodePrefixKey(groupingKey: Any): UnsafeRow = {
- val keyRow = keySerializer.apply(groupingKey)
- prefixKeyProjection.apply(InternalRow(keyRow))
+ val keyRow = keySerializer(groupingKey)
+ prefixKeyProjection(InternalRow(keyRow))
}
def decodePrefixKey(retUnsafeRow: UnsafeRow): Any = {
- keyDeserializer.apply(retUnsafeRow)
+ keyDeserializer(retUnsafeRow)
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
