You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ka...@apache.org on 2024/12/14 01:47:30 UTC
(spark) branch master updated: [SPARK-50526][SS] Add store encoding format conf into offset log and block non supported stateful operators from using avro
This is an automated email from the ASF dual-hosted git repository.
kabhwan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 3db46bf1fb48 [SPARK-50526][SS] Add store encoding format conf into offset log and block non supported stateful operators from using avro
3db46bf1fb48 is described below
commit 3db46bf1fb48e93d55b7f766e444e6d4a396981c
Author: Anish Shrigondekar <an...@databricks.com>
AuthorDate: Sat Dec 14 10:47:16 2024 +0900
[SPARK-50526][SS] Add store encoding format conf into offset log and block non supported stateful operators from using avro
### What changes were proposed in this pull request?
Add store encoding format conf into offset log and block non supported stateful operators from using avro
### Why are the changes needed?
Changes are needed to ensure that encoding format info is stored in offset log and that trying to use avro in a query that has stateful operators that don't support this encoding format yet will result in query failure.
### Does this PR introduce _any_ user-facing change?
Yes
### How was this patch tested?
Added unit tests
```
[info] Run completed in 2 seconds, 547 milliseconds.
[info] Total number of tests run: 6
[info] Suites: completed 1, aborted 0
[info] Tests: succeeded 6, failed 0, canceled 0, ignored 0, pending 0
[info] All tests passed.
[success] Total time: 13 s, completed Dec 9, 2024, 6:15:25 PM
```
### Was this patch authored or co-authored using generative AI tooling?
No
Closes #49121 from anishshri-db/task/SPARK-50526.
Authored-by: Anish Shrigondekar <an...@databricks.com>
Signed-off-by: Jungtaek Lim <ka...@gmail.com>
---
.../analysis/UnsupportedOperationChecker.scala | 39 +++++
.../spark/sql/execution/streaming/OffsetSeq.scala | 5 +-
.../commits/.0.crc | Bin 0 -> 12 bytes
.../commits/.1.crc | Bin 0 -> 12 bytes
.../checkpoint-version-4.0.0-tws-avro/commits/0 | 2 +
.../checkpoint-version-4.0.0-tws-avro/commits/1 | 2 +
.../checkpoint-version-4.0.0-tws-avro/metadata | 1 +
.../offsets/.0.crc | Bin 0 -> 16 bytes
.../offsets/.1.crc | Bin 0 -> 16 bytes
.../checkpoint-version-4.0.0-tws-avro/offsets/0 | 3 +
.../checkpoint-version-4.0.0-tws-avro/offsets/1 | 3 +
.../state/0/0/.1.changelog.crc | Bin 0 -> 12 bytes
.../state/0/0/.1.zip.crc | Bin 0 -> 40 bytes
.../state/0/0/.2.changelog.crc | Bin 0 -> 12 bytes
.../state/0/0/1.changelog | Bin 0 -> 63 bytes
.../state/0/0/1.zip | Bin 0 -> 3782 bytes
.../state/0/0/2.changelog | Bin 0 -> 63 bytes
...08-7b1ee246-6831-4c62-9fd7-7741cb534368.sst.crc | Bin 0 -> 20 bytes
...000008-7b1ee246-6831-4c62-9fd7-7741cb534368.sst | Bin 0 -> 1205 bytes
.../state/0/1/.1.changelog.crc | Bin 0 -> 12 bytes
.../state/0/1/.1.zip.crc | Bin 0 -> 40 bytes
.../state/0/1/.2.changelog.crc | Bin 0 -> 12 bytes
.../state/0/1/1.changelog | Bin 0 -> 47 bytes
.../state/0/1/1.zip | Bin 0 -> 3609 bytes
.../state/0/1/2.changelog | Bin 0 -> 63 bytes
.../state/0/2/.1.changelog.crc | Bin 0 -> 12 bytes
.../state/0/2/.1.zip.crc | Bin 0 -> 40 bytes
.../state/0/2/.2.changelog.crc | Bin 0 -> 12 bytes
.../state/0/2/1.changelog | Bin 0 -> 47 bytes
.../state/0/2/1.zip | Bin 0 -> 3610 bytes
.../state/0/2/2.changelog | Bin 0 -> 47 bytes
.../state/0/3/.1.changelog.crc | Bin 0 -> 12 bytes
.../state/0/3/.1.zip.crc | Bin 0 -> 40 bytes
.../state/0/3/.2.changelog.crc | Bin 0 -> 12 bytes
.../state/0/3/1.changelog | Bin 0 -> 47 bytes
.../state/0/3/1.zip | Bin 0 -> 3610 bytes
.../state/0/3/2.changelog | Bin 0 -> 47 bytes
.../state/0/4/.1.changelog.crc | Bin 0 -> 12 bytes
.../state/0/4/.1.zip.crc | Bin 0 -> 40 bytes
.../state/0/4/.2.changelog.crc | Bin 0 -> 12 bytes
.../state/0/4/1.changelog | Bin 0 -> 47 bytes
.../state/0/4/1.zip | Bin 0 -> 3610 bytes
.../state/0/4/2.changelog | Bin 0 -> 47 bytes
.../state/0/_metadata/v2/.0.crc | Bin 0 -> 16 bytes
.../state/0/_metadata/v2/0 | 2 +
.../.0_6b12d3c5-57e6-4001-8321-3ae63d6be7a0.crc | Bin 0 -> 12 bytes
.../default/0_6b12d3c5-57e6-4001-8321-3ae63d6be7a0 | Bin 0 -> 284 bytes
.../commits/.0.crc | Bin 0 -> 12 bytes
.../commits/.1.crc | Bin 0 -> 12 bytes
.../commits/0 | 2 +
.../commits/1 | 2 +
.../metadata | 1 +
.../offsets/.0.crc | Bin 0 -> 16 bytes
.../offsets/.1.crc | Bin 0 -> 16 bytes
.../offsets/0 | 3 +
.../offsets/1 | 3 +
.../state/0/0/.1.changelog.crc | Bin 0 -> 12 bytes
.../state/0/0/.1.zip.crc | Bin 0 -> 40 bytes
.../state/0/0/.2.changelog.crc | Bin 0 -> 12 bytes
.../state/0/0/1.changelog | Bin 0 -> 88 bytes
.../state/0/0/1.zip | Bin 0 -> 3785 bytes
.../state/0/0/2.changelog | Bin 0 -> 88 bytes
...08-9b6e23ce-e7de-4df8-b320-2b0378b53e52.sst.crc | Bin 0 -> 20 bytes
...000008-9b6e23ce-e7de-4df8-b320-2b0378b53e52.sst | Bin 0 -> 1228 bytes
.../state/0/1/.1.changelog.crc | Bin 0 -> 12 bytes
.../state/0/1/.1.zip.crc | Bin 0 -> 40 bytes
.../state/0/1/.2.changelog.crc | Bin 0 -> 12 bytes
.../state/0/1/1.changelog | Bin 0 -> 47 bytes
.../state/0/1/1.zip | Bin 0 -> 3610 bytes
.../state/0/1/2.changelog | Bin 0 -> 88 bytes
.../state/0/2/.1.changelog.crc | Bin 0 -> 12 bytes
.../state/0/2/.1.zip.crc | Bin 0 -> 40 bytes
.../state/0/2/.2.changelog.crc | Bin 0 -> 12 bytes
.../state/0/2/1.changelog | Bin 0 -> 47 bytes
.../state/0/2/1.zip | Bin 0 -> 3610 bytes
.../state/0/2/2.changelog | Bin 0 -> 47 bytes
.../state/0/3/.1.changelog.crc | Bin 0 -> 12 bytes
.../state/0/3/.1.zip.crc | Bin 0 -> 40 bytes
.../state/0/3/.2.changelog.crc | Bin 0 -> 12 bytes
.../state/0/3/1.changelog | Bin 0 -> 47 bytes
.../state/0/3/1.zip | Bin 0 -> 3610 bytes
.../state/0/3/2.changelog | Bin 0 -> 47 bytes
.../state/0/4/.1.changelog.crc | Bin 0 -> 12 bytes
.../state/0/4/.1.zip.crc | Bin 0 -> 40 bytes
.../state/0/4/.2.changelog.crc | Bin 0 -> 12 bytes
.../state/0/4/1.changelog | Bin 0 -> 47 bytes
.../state/0/4/1.zip | Bin 0 -> 3609 bytes
.../state/0/4/2.changelog | Bin 0 -> 47 bytes
.../state/0/_metadata/v2/.0.crc | Bin 0 -> 16 bytes
.../state/0/_metadata/v2/0 | 2 +
.../.0_2e8e6b52-e3c3-4184-b8ef-8d391b75d751.crc | Bin 0 -> 12 bytes
.../default/0_2e8e6b52-e3c3-4184-b8ef-8d391b75d751 | Bin 0 -> 284 bytes
.../execution/streaming/OffsetSeqLogSuite.scala | 67 +++++++
.../sql/streaming/StreamingAggregationSuite.scala | 20 +++
.../streaming/StreamingDeduplicationSuite.scala | 15 ++
.../TransformWithStateChainingSuite.scala | 195 ++++++++++++---------
96 files changed, 278 insertions(+), 89 deletions(-)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationChecker.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationChecker.scala
index 5b7583c763c0..f7ab41bd6f96 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationChecker.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationChecker.scala
@@ -17,6 +17,8 @@
package org.apache.spark.sql.catalyst.analysis
+import java.util.Locale
+
import org.apache.spark.internal.{Logging, MDC}
import org.apache.spark.internal.LogKeys.{ANALYSIS_ERROR, QUERY_PLAN}
import org.apache.spark.sql.AnalysisException
@@ -140,6 +142,38 @@ object UnsupportedOperationChecker extends Logging {
}
}
+ private def checkAvroSupportForStatefulOperator(p: LogicalPlan): Option[String] = p match {
+ // TODO: remove operators from this list as support for avro encoding is added
+ case s: Aggregate if s.isStreaming => Some("aggregation")
+ // Since the Distinct node will be replaced to Aggregate in the optimizer rule
+ // [[ReplaceDistinctWithAggregate]], here we also need to check all Distinct node by
+ // assuming it as Aggregate.
+ case d @ Distinct(_: LogicalPlan) if d.isStreaming => Some("distinct")
+ case _ @ Join(left, right, _, _, _) if left.isStreaming && right.isStreaming => Some("join")
+ case f: FlatMapGroupsWithState if f.isStreaming => Some("flatMapGroupsWithState")
+ case f: FlatMapGroupsInPandasWithState if f.isStreaming =>
+ Some("applyInPandasWithState")
+ case d: Deduplicate if d.isStreaming => Some("dropDuplicates")
+ case d: DeduplicateWithinWatermark if d.isStreaming => Some("dropDuplicatesWithinWatermark")
+ case _ => None
+ }
+
+ // Rule to check that avro encoding format is not supported in case any
+ // non-transformWithState stateful streaming operators are present in the query.
+ def checkSupportedStoreEncodingFormats(plan: LogicalPlan): Unit = {
+ val storeEncodingFormat = SQLConf.get.stateStoreEncodingFormat
+ if (storeEncodingFormat.toLowerCase(Locale.ROOT) == "avro") {
+ plan.foreach { subPlan =>
+ val operatorOpt = checkAvroSupportForStatefulOperator(subPlan)
+ if (operatorOpt.isDefined) {
+ val errorMsg = "State store encoding format as avro is not supported for " +
+ s"operator=${operatorOpt.get} used within the query"
+ throwError(errorMsg)(plan)
+ }
+ }
+ }
+ }
+
def checkForStreaming(plan: LogicalPlan, outputMode: OutputMode): Unit = {
if (!plan.isStreaming) {
throwError(
@@ -199,6 +233,11 @@ object UnsupportedOperationChecker extends Logging {
"DataFrames/Datasets")(plan)
}
+ // check to see that if store encoding format is set to true, then we have no stateful
+ // operators in the query or only variants of operators that support avro encoding such as
+ // transformWithState.
+ checkSupportedStoreEncodingFormats(plan)
+
val aggregates = collectStreamingAggregates(plan)
// Disallow some output mode
outputMode match {
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/OffsetSeq.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/OffsetSeq.scala
index e1e5b3a7ef88..a599f3bc6611 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/OffsetSeq.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/OffsetSeq.scala
@@ -102,7 +102,7 @@ object OffsetSeqMetadata extends Logging {
FLATMAPGROUPSWITHSTATE_STATE_FORMAT_VERSION, STREAMING_AGGREGATION_STATE_FORMAT_VERSION,
STREAMING_JOIN_STATE_FORMAT_VERSION, STATE_STORE_COMPRESSION_CODEC,
STATE_STORE_ROCKSDB_FORMAT_VERSION, STATEFUL_OPERATOR_USE_STRICT_DISTRIBUTION,
- PRUNE_FILTERS_CAN_PRUNE_STREAMING_SUBPLAN
+ PRUNE_FILTERS_CAN_PRUNE_STREAMING_SUBPLAN, STREAMING_STATE_STORE_ENCODING_FORMAT
)
/**
@@ -125,7 +125,8 @@ object OffsetSeqMetadata extends Logging {
SymmetricHashJoinStateManager.legacyVersion.toString,
STATE_STORE_COMPRESSION_CODEC.key -> CompressionCodec.LZ4,
STATEFUL_OPERATOR_USE_STRICT_DISTRIBUTION.key -> "false",
- PRUNE_FILTERS_CAN_PRUNE_STREAMING_SUBPLAN.key -> "true"
+ PRUNE_FILTERS_CAN_PRUNE_STREAMING_SUBPLAN.key -> "true",
+ STREAMING_STATE_STORE_ENCODING_FORMAT.key -> "unsaferow"
)
def apply(json: String): OffsetSeqMetadata = Serialization.read[OffsetSeqMetadata](json)
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/commits/.0.crc b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/commits/.0.crc
new file mode 100644
index 000000000000..dd09db7ad216
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/commits/.0.crc differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/commits/.1.crc b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/commits/.1.crc
new file mode 100644
index 000000000000..dd09db7ad216
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/commits/.1.crc differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/commits/0 b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/commits/0
new file mode 100644
index 000000000000..7e7f3b21c4e7
--- /dev/null
+++ b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/commits/0
@@ -0,0 +1,2 @@
+v1
+{"nextBatchWatermarkMs":0,"stateUniqueIds":{}}
\ No newline at end of file
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/commits/1 b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/commits/1
new file mode 100644
index 000000000000..7e7f3b21c4e7
--- /dev/null
+++ b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/commits/1
@@ -0,0 +1,2 @@
+v1
+{"nextBatchWatermarkMs":0,"stateUniqueIds":{}}
\ No newline at end of file
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/metadata b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/metadata
new file mode 100644
index 000000000000..6888983b0bc5
--- /dev/null
+++ b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/metadata
@@ -0,0 +1 @@
+{"id":"f3f30619-9175-4329-97a7-f5629deaad89"}
\ No newline at end of file
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/offsets/.0.crc b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/offsets/.0.crc
new file mode 100644
index 000000000000..400184017c91
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/offsets/.0.crc differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/offsets/.1.crc b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/offsets/.1.crc
new file mode 100644
index 000000000000..397dde18c6d5
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/offsets/.1.crc differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/offsets/0 b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/offsets/0
new file mode 100644
index 000000000000..8177241a333b
--- /dev/null
+++ b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/offsets/0
@@ -0,0 +1,3 @@
+v1
+{"batchWatermarkMs":0,"batchTimestampMs":1734074255407,"conf":{"spark.sql.streaming.stateStore.providerClass":"org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider","spark.sql.streaming.stateStore.rocksdb.formatVersion":"5","spark.sql.streaming.stateStore.encodingFormat":"avro","spark.sql.streaming.statefulOperator.useStrictDistribution":"true","spark.sql.streaming.flatMapGroupsWithState.stateFormatVersion":"2","spark.sql.streaming.aggregation.stateFormatVersion":"2", [...]
+0
\ No newline at end of file
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/offsets/1 b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/offsets/1
new file mode 100644
index 000000000000..cf51e39873cd
--- /dev/null
+++ b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/offsets/1
@@ -0,0 +1,3 @@
+v1
+{"batchWatermarkMs":0,"batchTimestampMs":1734074257473,"conf":{"spark.sql.streaming.stateStore.providerClass":"org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider","spark.sql.streaming.stateStore.rocksdb.formatVersion":"5","spark.sql.streaming.stateStore.encodingFormat":"avro","spark.sql.streaming.statefulOperator.useStrictDistribution":"true","spark.sql.streaming.flatMapGroupsWithState.stateFormatVersion":"2","spark.sql.streaming.aggregation.stateFormatVersion":"2", [...]
+1
\ No newline at end of file
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/0/.1.changelog.crc b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/0/.1.changelog.crc
new file mode 100644
index 000000000000..d03e8f51a6f7
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/0/.1.changelog.crc differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/0/.1.zip.crc b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/0/.1.zip.crc
new file mode 100644
index 000000000000..9fe6838ba35e
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/0/.1.zip.crc differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/0/.2.changelog.crc b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/0/.2.changelog.crc
new file mode 100644
index 000000000000..c790833be99a
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/0/.2.changelog.crc differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/0/1.changelog b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/0/1.changelog
new file mode 100644
index 000000000000..a579fe940633
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/0/1.changelog differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/0/1.zip b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/0/1.zip
new file mode 100644
index 000000000000..1f2b15b840c2
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/0/1.zip differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/0/2.changelog b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/0/2.changelog
new file mode 100644
index 000000000000..6df7672a3d0e
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/0/2.changelog differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/0/SSTs/.000008-7b1ee246-6831-4c62-9fd7-7741cb534368.sst.crc b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/0/SSTs/.000008-7b1ee246-6831-4c62-9fd7-7741cb534368.sst.crc
new file mode 100644
index 000000000000..2026a6dcab3b
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/0/SSTs/.000008-7b1ee246-6831-4c62-9fd7-7741cb534368.sst.crc differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/0/SSTs/000008-7b1ee246-6831-4c62-9fd7-7741cb534368.sst b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/0/SSTs/000008-7b1ee246-6831-4c62-9fd7-7741cb534368.sst
new file mode 100644
index 000000000000..c4f2175a4737
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/0/SSTs/000008-7b1ee246-6831-4c62-9fd7-7741cb534368.sst differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/1/.1.changelog.crc b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/1/.1.changelog.crc
new file mode 100644
index 000000000000..22e87bcdbe20
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/1/.1.changelog.crc differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/1/.1.zip.crc b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/1/.1.zip.crc
new file mode 100644
index 000000000000..0b52e7b4922a
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/1/.1.zip.crc differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/1/.2.changelog.crc b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/1/.2.changelog.crc
new file mode 100644
index 000000000000..889a96a47f0a
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/1/.2.changelog.crc differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/1/1.changelog b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/1/1.changelog
new file mode 100644
index 000000000000..85a6a13b976f
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/1/1.changelog differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/1/1.zip b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/1/1.zip
new file mode 100644
index 000000000000..3986c42e7e5e
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/1/1.zip differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/1/2.changelog b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/1/2.changelog
new file mode 100644
index 000000000000..13ec6c9cdd84
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/1/2.changelog differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/2/.1.changelog.crc b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/2/.1.changelog.crc
new file mode 100644
index 000000000000..22e87bcdbe20
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/2/.1.changelog.crc differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/2/.1.zip.crc b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/2/.1.zip.crc
new file mode 100644
index 000000000000..6e5b8098e6e4
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/2/.1.zip.crc differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/2/.2.changelog.crc b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/2/.2.changelog.crc
new file mode 100644
index 000000000000..22e87bcdbe20
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/2/.2.changelog.crc differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/2/1.changelog b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/2/1.changelog
new file mode 100644
index 000000000000..85a6a13b976f
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/2/1.changelog differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/2/1.zip b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/2/1.zip
new file mode 100644
index 000000000000..59318e5f734e
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/2/1.zip differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/2/2.changelog b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/2/2.changelog
new file mode 100644
index 000000000000..85a6a13b976f
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/2/2.changelog differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/3/.1.changelog.crc b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/3/.1.changelog.crc
new file mode 100644
index 000000000000..22e87bcdbe20
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/3/.1.changelog.crc differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/3/.1.zip.crc b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/3/.1.zip.crc
new file mode 100644
index 000000000000..4899bd7e696c
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/3/.1.zip.crc differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/3/.2.changelog.crc b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/3/.2.changelog.crc
new file mode 100644
index 000000000000..22e87bcdbe20
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/3/.2.changelog.crc differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/3/1.changelog b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/3/1.changelog
new file mode 100644
index 000000000000..85a6a13b976f
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/3/1.changelog differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/3/1.zip b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/3/1.zip
new file mode 100644
index 000000000000..29a07a94b5db
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/3/1.zip differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/3/2.changelog b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/3/2.changelog
new file mode 100644
index 000000000000..85a6a13b976f
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/3/2.changelog differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/4/.1.changelog.crc b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/4/.1.changelog.crc
new file mode 100644
index 000000000000..22e87bcdbe20
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/4/.1.changelog.crc differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/4/.1.zip.crc b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/4/.1.zip.crc
new file mode 100644
index 000000000000..846c94b6dfcf
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/4/.1.zip.crc differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/4/.2.changelog.crc b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/4/.2.changelog.crc
new file mode 100644
index 000000000000..22e87bcdbe20
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/4/.2.changelog.crc differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/4/1.changelog b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/4/1.changelog
new file mode 100644
index 000000000000..85a6a13b976f
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/4/1.changelog differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/4/1.zip b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/4/1.zip
new file mode 100644
index 000000000000..38e6a7581458
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/4/1.zip differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/4/2.changelog b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/4/2.changelog
new file mode 100644
index 000000000000..85a6a13b976f
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/4/2.changelog differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/_metadata/v2/.0.crc b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/_metadata/v2/.0.crc
new file mode 100644
index 000000000000..9850f2cdfc9b
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/_metadata/v2/.0.crc differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/_metadata/v2/0 b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/_metadata/v2/0
new file mode 100644
index 000000000000..5c5303653046
--- /dev/null
+++ b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/_metadata/v2/0
@@ -0,0 +1,2 @@
+v2
+{"operatorInfo":{"operatorId":0,"operatorName":"transformWithStateExec"},"stateStoreInfo":[{"storeName":"default","numColsPrefixKey":0,"numPartitions":5,"stateSchemaFilePath":"file:/Users/anish.shrigondekar/spark/spark/target/tmp/spark-dcaeba6f-ff09-4f91-ba1b-4d14fe53cc9f/state/0/_stateSchema/default/0_6b12d3c5-57e6-4001-8321-3ae63d6be7a0"}],"operatorPropertiesJson":"{\"timeMode\":\"NoTime\",\"outputMode\":\"Update\",\"stateVariables\":[{\"stateName\":\"countState\",\"stateVariableType\" [...]
\ No newline at end of file
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/_stateSchema/default/.0_6b12d3c5-57e6-4001-8321-3ae63d6be7a0.crc b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/_stateSchema/default/.0_6b12d3c5-57e6-4001-8321-3ae63d6be7a0.crc
new file mode 100644
index 000000000000..3c16c8244a3b
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/_stateSchema/default/.0_6b12d3c5-57e6-4001-8321-3ae63d6be7a0.crc differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/_stateSchema/default/0_6b12d3c5-57e6-4001-8321-3ae63d6be7a0 b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/_stateSchema/default/0_6b12d3c5-57e6-4001-8321-3ae63d6be7a0
new file mode 100644
index 000000000000..cd3e8f6d96bf
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-avro/state/0/_stateSchema/default/0_6b12d3c5-57e6-4001-8321-3ae63d6be7a0 differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/commits/.0.crc b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/commits/.0.crc
new file mode 100644
index 000000000000..dd09db7ad216
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/commits/.0.crc differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/commits/.1.crc b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/commits/.1.crc
new file mode 100644
index 000000000000..dd09db7ad216
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/commits/.1.crc differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/commits/0 b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/commits/0
new file mode 100644
index 000000000000..7e7f3b21c4e7
--- /dev/null
+++ b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/commits/0
@@ -0,0 +1,2 @@
+v1
+{"nextBatchWatermarkMs":0,"stateUniqueIds":{}}
\ No newline at end of file
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/commits/1 b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/commits/1
new file mode 100644
index 000000000000..7e7f3b21c4e7
--- /dev/null
+++ b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/commits/1
@@ -0,0 +1,2 @@
+v1
+{"nextBatchWatermarkMs":0,"stateUniqueIds":{}}
\ No newline at end of file
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/metadata b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/metadata
new file mode 100644
index 000000000000..d23698154575
--- /dev/null
+++ b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/metadata
@@ -0,0 +1 @@
+{"id":"1341f9d1-5100-4426-876c-2754aeaca02b"}
\ No newline at end of file
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/offsets/.0.crc b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/offsets/.0.crc
new file mode 100644
index 000000000000..15e5afc45b88
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/offsets/.0.crc differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/offsets/.1.crc b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/offsets/.1.crc
new file mode 100644
index 000000000000..de2378b2b16e
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/offsets/.1.crc differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/offsets/0 b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/offsets/0
new file mode 100644
index 000000000000..2f2a25bda632
--- /dev/null
+++ b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/offsets/0
@@ -0,0 +1,3 @@
+v1
+{"batchWatermarkMs":0,"batchTimestampMs":1734074067729,"conf":{"spark.sql.streaming.stateStore.providerClass":"org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider","spark.sql.streaming.stateStore.rocksdb.formatVersion":"5","spark.sql.streaming.stateStore.encodingFormat":"unsaferow","spark.sql.streaming.statefulOperator.useStrictDistribution":"true","spark.sql.streaming.flatMapGroupsWithState.stateFormatVersion":"2","spark.sql.streaming.aggregation.stateFormatVersion" [...]
+0
\ No newline at end of file
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/offsets/1 b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/offsets/1
new file mode 100644
index 000000000000..3295f1a1579b
--- /dev/null
+++ b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/offsets/1
@@ -0,0 +1,3 @@
+v1
+{"batchWatermarkMs":0,"batchTimestampMs":1734074071551,"conf":{"spark.sql.streaming.stateStore.providerClass":"org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider","spark.sql.streaming.stateStore.rocksdb.formatVersion":"5","spark.sql.streaming.stateStore.encodingFormat":"unsaferow","spark.sql.streaming.statefulOperator.useStrictDistribution":"true","spark.sql.streaming.flatMapGroupsWithState.stateFormatVersion":"2","spark.sql.streaming.aggregation.stateFormatVersion" [...]
+1
\ No newline at end of file
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/0/.1.changelog.crc b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/0/.1.changelog.crc
new file mode 100644
index 000000000000..a1ca2c3ed5c1
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/0/.1.changelog.crc differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/0/.1.zip.crc b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/0/.1.zip.crc
new file mode 100644
index 000000000000..ad0d75698608
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/0/.1.zip.crc differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/0/.2.changelog.crc b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/0/.2.changelog.crc
new file mode 100644
index 000000000000..4ad793ade578
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/0/.2.changelog.crc differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/0/1.changelog b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/0/1.changelog
new file mode 100644
index 000000000000..21cc1e7055f4
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/0/1.changelog differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/0/1.zip b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/0/1.zip
new file mode 100644
index 000000000000..2c6fbb713b43
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/0/1.zip differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/0/2.changelog b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/0/2.changelog
new file mode 100644
index 000000000000..2375a971fdc2
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/0/2.changelog differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/0/SSTs/.000008-9b6e23ce-e7de-4df8-b320-2b0378b53e52.sst.crc b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/0/SSTs/.000008-9b6e23ce-e7de-4df8-b320-2b0378b53e52.sst.crc
new file mode 100644
index 000000000000..d72d0acb543e
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/0/SSTs/.000008-9b6e23ce-e7de-4df8-b320-2b0378b53e52.sst.crc differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/0/SSTs/000008-9b6e23ce-e7de-4df8-b320-2b0378b53e52.sst b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/0/SSTs/000008-9b6e23ce-e7de-4df8-b320-2b0378b53e52.sst
new file mode 100644
index 000000000000..bed4218de7ec
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/0/SSTs/000008-9b6e23ce-e7de-4df8-b320-2b0378b53e52.sst differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/1/.1.changelog.crc b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/1/.1.changelog.crc
new file mode 100644
index 000000000000..22e87bcdbe20
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/1/.1.changelog.crc differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/1/.1.zip.crc b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/1/.1.zip.crc
new file mode 100644
index 000000000000..2e4c44069438
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/1/.1.zip.crc differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/1/.2.changelog.crc b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/1/.2.changelog.crc
new file mode 100644
index 000000000000..57fb3ea9d4a6
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/1/.2.changelog.crc differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/1/1.changelog b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/1/1.changelog
new file mode 100644
index 000000000000..85a6a13b976f
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/1/1.changelog differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/1/1.zip b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/1/1.zip
new file mode 100644
index 000000000000..85388bed9dc4
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/1/1.zip differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/1/2.changelog b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/1/2.changelog
new file mode 100644
index 000000000000..daf1c187e8b7
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/1/2.changelog differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/2/.1.changelog.crc b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/2/.1.changelog.crc
new file mode 100644
index 000000000000..22e87bcdbe20
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/2/.1.changelog.crc differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/2/.1.zip.crc b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/2/.1.zip.crc
new file mode 100644
index 000000000000..03b4665cea14
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/2/.1.zip.crc differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/2/.2.changelog.crc b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/2/.2.changelog.crc
new file mode 100644
index 000000000000..22e87bcdbe20
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/2/.2.changelog.crc differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/2/1.changelog b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/2/1.changelog
new file mode 100644
index 000000000000..85a6a13b976f
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/2/1.changelog differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/2/1.zip b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/2/1.zip
new file mode 100644
index 000000000000..bf6c8277bc19
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/2/1.zip differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/2/2.changelog b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/2/2.changelog
new file mode 100644
index 000000000000..85a6a13b976f
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/2/2.changelog differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/3/.1.changelog.crc b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/3/.1.changelog.crc
new file mode 100644
index 000000000000..22e87bcdbe20
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/3/.1.changelog.crc differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/3/.1.zip.crc b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/3/.1.zip.crc
new file mode 100644
index 000000000000..a28994f96def
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/3/.1.zip.crc differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/3/.2.changelog.crc b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/3/.2.changelog.crc
new file mode 100644
index 000000000000..22e87bcdbe20
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/3/.2.changelog.crc differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/3/1.changelog b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/3/1.changelog
new file mode 100644
index 000000000000..85a6a13b976f
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/3/1.changelog differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/3/1.zip b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/3/1.zip
new file mode 100644
index 000000000000..f4b734fa3695
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/3/1.zip differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/3/2.changelog b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/3/2.changelog
new file mode 100644
index 000000000000..85a6a13b976f
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/3/2.changelog differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/4/.1.changelog.crc b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/4/.1.changelog.crc
new file mode 100644
index 000000000000..22e87bcdbe20
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/4/.1.changelog.crc differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/4/.1.zip.crc b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/4/.1.zip.crc
new file mode 100644
index 000000000000..80ad867da34f
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/4/.1.zip.crc differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/4/.2.changelog.crc b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/4/.2.changelog.crc
new file mode 100644
index 000000000000..22e87bcdbe20
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/4/.2.changelog.crc differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/4/1.changelog b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/4/1.changelog
new file mode 100644
index 000000000000..85a6a13b976f
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/4/1.changelog differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/4/1.zip b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/4/1.zip
new file mode 100644
index 000000000000..e91c9a674161
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/4/1.zip differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/4/2.changelog b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/4/2.changelog
new file mode 100644
index 000000000000..85a6a13b976f
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/4/2.changelog differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/_metadata/v2/.0.crc b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/_metadata/v2/.0.crc
new file mode 100644
index 000000000000..257c878a1611
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/_metadata/v2/.0.crc differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/_metadata/v2/0 b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/_metadata/v2/0
new file mode 100644
index 000000000000..313c13df69ac
--- /dev/null
+++ b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/_metadata/v2/0
@@ -0,0 +1,2 @@
+v2
+{"operatorInfo":{"operatorId":0,"operatorName":"transformWithStateExec"},"stateStoreInfo":[{"storeName":"default","numColsPrefixKey":0,"numPartitions":5,"stateSchemaFilePath":"file:/Users/anish.shrigondekar/spark/spark/target/tmp/spark-ae28252a-e696-4653-a9a5-7a9a0766f4c1/state/0/_stateSchema/default/0_2e8e6b52-e3c3-4184-b8ef-8d391b75d751"}],"operatorPropertiesJson":"{\"timeMode\":\"NoTime\",\"outputMode\":\"Update\",\"stateVariables\":[{\"stateName\":\"countState\",\"stateVariableType\" [...]
\ No newline at end of file
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/_stateSchema/default/.0_2e8e6b52-e3c3-4184-b8ef-8d391b75d751.crc b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/_stateSchema/default/.0_2e8e6b52-e3c3-4184-b8ef-8d391b75d751.crc
new file mode 100644
index 000000000000..3c16c8244a3b
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/_stateSchema/default/.0_2e8e6b52-e3c3-4184-b8ef-8d391b75d751.crc differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/_stateSchema/default/0_2e8e6b52-e3c3-4184-b8ef-8d391b75d751 b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/_stateSchema/default/0_2e8e6b52-e3c3-4184-b8ef-8d391b75d751
new file mode 100644
index 000000000000..cd3e8f6d96bf
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.0.0-tws-unsaferow/state/0/_stateSchema/default/0_2e8e6b52-e3c3-4184-b8ef-8d391b75d751 differ
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/OffsetSeqLogSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/OffsetSeqLogSuite.scala
index e5dfa3316490..5f7205eaf4bb 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/OffsetSeqLogSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/OffsetSeqLogSuite.scala
@@ -19,6 +19,8 @@ package org.apache.spark.sql.execution.streaming
import java.io.File
+import org.apache.commons.io.FileUtils
+
import org.apache.spark.sql.catalyst.util.stringToFile
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
@@ -129,4 +131,69 @@ class OffsetSeqLogSuite extends SharedSparkSession {
val log = new OffsetSeqLog(spark, input.toString)
log.getLatest().get
}
+
+ // SPARK-50526 - sanity tests to ensure that values are set correctly for state store
+ // encoding format within OffsetSeqMetadata
+ test("offset log records defaults to unsafeRow for store encoding format") {
+ val offsetSeqMetadata = OffsetSeqMetadata.apply(batchWatermarkMs = 0, batchTimestampMs = 0,
+ spark.conf)
+ assert(offsetSeqMetadata.conf.get(SQLConf.STREAMING_STATE_STORE_ENCODING_FORMAT.key) ===
+ Some("unsaferow"))
+ }
+
+ test("offset log uses the store encoding format set in the conf") {
+ val offsetSeqMetadata = OffsetSeqMetadata.apply(batchWatermarkMs = 0, batchTimestampMs = 0,
+ Map(SQLConf.STREAMING_STATE_STORE_ENCODING_FORMAT.key -> "avro"))
+ assert(offsetSeqMetadata.conf.get(SQLConf.STREAMING_STATE_STORE_ENCODING_FORMAT.key) ===
+ Some("avro"))
+ }
+
+ // Verify whether entry exists within the offset log and has the right value or that we pick up
+ // the correct default values when populating the session conf.
+ private def verifyOffsetLogEntry(
+ checkpointDir: String,
+ entryExists: Boolean,
+ encodingFormat: String): Unit = {
+ val log = new OffsetSeqLog(spark, s"$checkpointDir/offsets")
+ val latestBatchId = log.getLatestBatchId()
+ assert(latestBatchId.isDefined, "No offset log entries found in the checkpoint location")
+
+ // Read the latest offset log
+ val offsetSeq = log.get(latestBatchId.get).get
+ val offsetSeqMetadata = offsetSeq.metadata.get
+
+ if (entryExists) {
+ val encodingFormatOpt = offsetSeqMetadata.conf.get(
+ SQLConf.STREAMING_STATE_STORE_ENCODING_FORMAT.key)
+ assert(encodingFormatOpt.isDefined, "No store encoding format found in the offset log entry")
+ assert(encodingFormatOpt.get == encodingFormat)
+ }
+
+ val clonedSqlConf = spark.sessionState.conf.clone()
+ OffsetSeqMetadata.setSessionConf(offsetSeqMetadata, clonedSqlConf)
+ assert(clonedSqlConf.stateStoreEncodingFormat == encodingFormat)
+ }
+
+ // verify that checkpoint created with different store encoding formats are read correctly
+ Seq("unsaferow", "avro").foreach { storeEncodingFormat =>
+ test(s"verify format values from checkpoint loc - $storeEncodingFormat") {
+ withTempDir { checkpointDir =>
+ val resourceUri = this.getClass.getResource(
+ "/structured-streaming/checkpoint-version-4.0.0-tws-" + storeEncodingFormat + "/").toURI
+ FileUtils.copyDirectory(new File(resourceUri), checkpointDir.getCanonicalFile)
+ verifyOffsetLogEntry(checkpointDir.getAbsolutePath, entryExists = true,
+ storeEncodingFormat)
+ }
+ }
+ }
+
+ test("verify format values from old checkpoint with Spark version 3.5.1") {
+ withTempDir { checkpointDir =>
+ val resourceUri = this.getClass.getResource(
+ "/structured-streaming/checkpoint-version-3.5.1-streaming-deduplication/").toURI
+ FileUtils.copyDirectory(new File(resourceUri), checkpointDir.getCanonicalFile)
+ verifyOffsetLogEntry(checkpointDir.getAbsolutePath, entryExists = false,
+ "unsaferow")
+ }
+ }
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingAggregationSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingAggregationSuite.scala
index bcf0d4ac4665..0f382f4ed77d 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingAggregationSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingAggregationSuite.scala
@@ -874,6 +874,26 @@ class StreamingAggregationSuite extends StateStoreMetricsTest with Assertions {
)
}
+ testWithAllStateVersions("test that avro encoding is not supported") {
+ val inputData = MemoryStream[Int]
+
+ val aggregated =
+ inputData.toDF()
+ .groupBy($"value")
+ .agg(count("*"))
+ .as[(Int, Long)]
+
+ val ex = intercept[Exception] {
+ withSQLConf(SQLConf.STREAMING_STATE_STORE_ENCODING_FORMAT.key -> "avro") {
+ testStream(aggregated, Update)(
+ AddData(inputData, 3),
+ ProcessAllAvailable()
+ )
+ }
+ }
+ assert(ex.getMessage.contains("State store encoding format as avro is not supported"))
+ }
+
private def prepareTestForChangingSchemaOfState(
tempDir: File): (MemoryStream[Int], DataFrame) = {
val inputData = MemoryStream[Int]
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingDeduplicationSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingDeduplicationSuite.scala
index ab9df9a1e5a6..040b99e55cb0 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingDeduplicationSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingDeduplicationSuite.scala
@@ -574,6 +574,21 @@ class StreamingDeduplicationSuite extends StateStoreMetricsTest {
matchPVals = true
)
}
+
+ test("test that avro encoding is not supported") {
+ val inputData = MemoryStream[String]
+ val result = inputData.toDS().dropDuplicates()
+
+ val ex = intercept[Exception] {
+ withSQLConf(SQLConf.STREAMING_STATE_STORE_ENCODING_FORMAT.key -> "avro") {
+ testStream(result, Append)(
+ AddData(inputData, "a"),
+ ProcessAllAvailable()
+ )
+ }
+ }
+ assert(ex.getMessage.contains("State store encoding format as avro is not supported"))
+ }
}
@SlowSQLTest
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/TransformWithStateChainingSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/TransformWithStateChainingSuite.scala
index 6888fcba45f3..0e963bec41b4 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/streaming/TransformWithStateChainingSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/streaming/TransformWithStateChainingSuite.scala
@@ -24,7 +24,7 @@ import org.apache.spark.{SparkRuntimeException, SparkThrowable}
import org.apache.spark.sql.AnalysisException
import org.apache.spark.sql.catalyst.ExtendedAnalysisException
import org.apache.spark.sql.execution.streaming.{MemoryStream, StreamExecution}
-import org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider
+import org.apache.spark.sql.execution.streaming.state.{AlsoTestWithEncodingTypes, AlsoTestWithRocksDBFeatures, RocksDBStateStoreProvider}
import org.apache.spark.sql.functions.window
import org.apache.spark.sql.internal.SQLConf
@@ -103,47 +103,53 @@ case class AggEventRow(
window: Window,
count: Long)
-class TransformWithStateChainingSuite extends StreamTest {
+class TransformWithStateChainingSuite extends StreamTest
+ with AlsoTestWithRocksDBFeatures
+ with AlsoTestWithEncodingTypes {
import testImplicits._
+ private def isAvroEnabled: Boolean = SQLConf.get.stateStoreEncodingFormat == "avro"
+
test("watermark is propagated correctly for next stateful operator" +
" after transformWithState") {
- withSQLConf(SQLConf.STATE_STORE_PROVIDER_CLASS.key ->
- classOf[RocksDBStateStoreProvider].getName) {
- val inputData = MemoryStream[InputEventRow]
+ if (!isAvroEnabled) {
+ withSQLConf(SQLConf.STATE_STORE_PROVIDER_CLASS.key ->
+ classOf[RocksDBStateStoreProvider].getName) {
+ val inputData = MemoryStream[InputEventRow]
- val result = inputData.toDS()
- .withWatermark("eventTime", "1 minute")
- .groupByKey(x => x.key)
- .transformWithState[OutputRow](
- new TestStatefulProcessor(),
- "outputEventTime",
- OutputMode.Append())
- .groupBy(window($"outputEventTime", "1 minute"))
- .count()
- .as[AggEventRow]
+ val result = inputData.toDS()
+ .withWatermark("eventTime", "1 minute")
+ .groupByKey(x => x.key)
+ .transformWithState[OutputRow](
+ new TestStatefulProcessor(),
+ "outputEventTime",
+ OutputMode.Append())
+ .groupBy(window($"outputEventTime", "1 minute"))
+ .count()
+ .as[AggEventRow]
- testStream(result, OutputMode.Append())(
- AddData(inputData, InputEventRow("k1", timestamp("2024-01-01 00:00:00"), "e1")),
- // watermark should be 1 minute behind `2024-01-01 00:00:00`, nothing is
- // emitted as all records have timestamp > epoch
- CheckNewAnswer(),
- Execute("assertWatermarkEquals") { q =>
- assertWatermarkEquals(q, timestamp("2023-12-31 23:59:00"))
- },
- AddData(inputData, InputEventRow("k1", timestamp("2024-02-01 00:00:00"), "e1")),
- // global watermark should now be 1 minute behind `2024-02-01 00:00:00`.
- CheckNewAnswer(AggEventRow(
- Window(timestamp("2024-01-01 00:00:00"), timestamp("2024-01-01 00:01:00")), 1)
- ),
- Execute("assertWatermarkEquals") { q =>
- assertWatermarkEquals(q, timestamp("2024-01-31 23:59:00"))
- },
- AddData(inputData, InputEventRow("k1", timestamp("2024-02-02 00:00:00"), "e1")),
- CheckNewAnswer(AggEventRow(
- Window(timestamp("2024-02-01 00:00:00"), timestamp("2024-02-01 00:01:00")), 1)
+ testStream(result, OutputMode.Append())(
+ AddData(inputData, InputEventRow("k1", timestamp("2024-01-01 00:00:00"), "e1")),
+ // watermark should be 1 minute behind `2024-01-01 00:00:00`, nothing is
+ // emitted as all records have timestamp > epoch
+ CheckNewAnswer(),
+ Execute("assertWatermarkEquals") { q =>
+ assertWatermarkEquals(q, timestamp("2023-12-31 23:59:00"))
+ },
+ AddData(inputData, InputEventRow("k1", timestamp("2024-02-01 00:00:00"), "e1")),
+ // global watermark should now be 1 minute behind `2024-02-01 00:00:00`.
+ CheckNewAnswer(AggEventRow(
+ Window(timestamp("2024-01-01 00:00:00"), timestamp("2024-01-01 00:01:00")), 1)
+ ),
+ Execute("assertWatermarkEquals") { q =>
+ assertWatermarkEquals(q, timestamp("2024-01-31 23:59:00"))
+ },
+ AddData(inputData, InputEventRow("k1", timestamp("2024-02-02 00:00:00"), "e1")),
+ CheckNewAnswer(AggEventRow(
+ Window(timestamp("2024-02-01 00:00:00"), timestamp("2024-02-01 00:01:00")), 1)
+ )
)
- )
+ }
}
}
@@ -166,33 +172,35 @@ class TransformWithStateChainingSuite extends StreamTest {
}
}
- test("missing eventTime column to transformWithState fails the query if" +
- " another stateful operator is added") {
- withSQLConf(SQLConf.STATE_STORE_PROVIDER_CLASS.key ->
- classOf[RocksDBStateStoreProvider].getName) {
- val inputData = MemoryStream[InputEventRow]
+ test("missing eventTime column to transformWithState fails the query if " +
+ "another stateful operator is added") {
+ if (!isAvroEnabled) {
+ withSQLConf(SQLConf.STATE_STORE_PROVIDER_CLASS.key ->
+ classOf[RocksDBStateStoreProvider].getName) {
+ val inputData = MemoryStream[InputEventRow]
- val result = inputData.toDS()
- .withWatermark("eventTime", "1 minute")
- .groupByKey(x => x.key)
- .transformWithState[OutputRow](
- new TestStatefulProcessor(),
- TimeMode.None(),
- OutputMode.Append())
- .groupBy(window($"outputEventTime", "1 minute"))
- .count()
-
- checkError(
- exception = intercept[AnalysisException] {
- testStream(result, OutputMode.Append())(
- StartStream()
- )
- },
- condition = "STREAMING_OUTPUT_MODE.UNSUPPORTED_OPERATION",
- sqlState = "42KDE",
- parameters = Map(
- "outputMode" -> "append",
- "operation" -> "streaming aggregations without watermark"))
+ val result = inputData.toDS()
+ .withWatermark("eventTime", "1 minute")
+ .groupByKey(x => x.key)
+ .transformWithState[OutputRow](
+ new TestStatefulProcessor(),
+ TimeMode.None(),
+ OutputMode.Append())
+ .groupBy(window($"outputEventTime", "1 minute"))
+ .count()
+
+ checkError(
+ exception = intercept[AnalysisException] {
+ testStream(result, OutputMode.Append())(
+ StartStream()
+ )
+ },
+ condition = "STREAMING_OUTPUT_MODE.UNSUPPORTED_OPERATION",
+ sqlState = "42KDE",
+ parameters = Map(
+ "outputMode" -> "append",
+ "operation" -> "streaming aggregations without watermark"))
+ }
}
}
@@ -234,25 +242,27 @@ class TransformWithStateChainingSuite extends StreamTest {
test("dropDuplicateWithWatermark after transformWithState operator" +
" fails if watermark column is not provided") {
- withSQLConf(SQLConf.STATE_STORE_PROVIDER_CLASS.key ->
- classOf[RocksDBStateStoreProvider].getName) {
- val inputData = MemoryStream[InputEventRow]
- val result = inputData.toDS()
- .withWatermark("eventTime", "1 minute")
- .groupByKey(x => x.key)
- .transformWithState[OutputRow](
- new TestStatefulProcessor(),
- TimeMode.None(),
- OutputMode.Append())
- .dropDuplicatesWithinWatermark()
+ if (!isAvroEnabled) {
+ withSQLConf(SQLConf.STATE_STORE_PROVIDER_CLASS.key ->
+ classOf[RocksDBStateStoreProvider].getName) {
+ val inputData = MemoryStream[InputEventRow]
+ val result = inputData.toDS()
+ .withWatermark("eventTime", "1 minute")
+ .groupByKey(x => x.key)
+ .transformWithState[OutputRow](
+ new TestStatefulProcessor(),
+ TimeMode.None(),
+ OutputMode.Append())
+ .dropDuplicatesWithinWatermark()
- val ex = intercept[ExtendedAnalysisException] {
- testStream(result, OutputMode.Append())(
- StartStream()
- )
+ val ex = intercept[ExtendedAnalysisException] {
+ testStream(result, OutputMode.Append())(
+ StartStream()
+ )
+ }
+ assert(ex.getMessage.contains("dropDuplicatesWithinWatermark is not supported on" +
+ " streaming DataFrames/DataSets without watermark"))
}
- assert(ex.getMessage.contains("dropDuplicatesWithinWatermark is not supported on" +
- " streaming DataFrames/DataSets without watermark"))
}
}
@@ -269,14 +279,25 @@ class TransformWithStateChainingSuite extends StreamTest {
OutputMode.Append())
.dropDuplicatesWithinWatermark()
- testStream(result, OutputMode.Append())(
- AddData(inputData, InputEventRow("k1", timestamp("2024-02-01 00:00:00"), "e1"),
- InputEventRow("k1", timestamp("2024-02-01 00:00:00"), "e1")),
- CheckNewAnswer(OutputRow("k1", timestamp("2024-02-01 00:00:00"), 2)),
- Execute("assertWatermarkEquals") { q =>
- assertWatermarkEquals(q, timestamp("2024-01-31 23:59:00"))
+ if (!isAvroEnabled) {
+ testStream(result, OutputMode.Append())(
+ AddData(inputData, InputEventRow("k1", timestamp("2024-02-01 00:00:00"), "e1"),
+ InputEventRow("k1", timestamp("2024-02-01 00:00:00"), "e1")),
+ CheckNewAnswer(OutputRow("k1", timestamp("2024-02-01 00:00:00"), 2)),
+ Execute("assertWatermarkEquals") { q =>
+ assertWatermarkEquals(q, timestamp("2024-01-31 23:59:00"))
+ }
+ )
+ } else {
+ val ex = intercept[Exception] {
+ testStream(result, OutputMode.Append())(
+ AddData(inputData, InputEventRow("k1", timestamp("2024-02-01 00:00:00"), "e1"),
+ InputEventRow("k1", timestamp("2024-02-01 00:00:00"), "e1")),
+ ProcessAllAvailable()
+ )
}
- )
+ assert(ex.getMessage.contains("State store encoding format as avro is not supported"))
+ }
}
}
@@ -340,7 +361,7 @@ class TransformWithStateChainingSuite extends StreamTest {
val inputData = MemoryStream[InputEventRow]
inputData.toDS()
.withWatermark("eventTime", "1 minute")
- .createTempView("tempViewWithWatermark")
+ .createOrReplaceTempView("tempViewWithWatermark")
val result = spark.readStream.table("tempViewWithWatermark")
.as[InputEventRow]
@@ -365,7 +386,7 @@ class TransformWithStateChainingSuite extends StreamTest {
classOf[RocksDBStateStoreProvider].getName) {
val inputData = MemoryStream[InputEventRow]
inputData.toDS()
- .createTempView("tempViewWithoutWatermark")
+ .createOrReplaceTempView("tempViewWithoutWatermark")
val ex = intercept[AnalysisException] {
val result = spark.readStream.table("tempViewWithoutWatermark")
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org