You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "cloud-fan (via GitHub)" <gi...@apache.org> on 2026/02/03 05:27:51 UTC
[PR] [SPARK-53656][SS][FOLLOWUP] Improve MemoryStream backward compatibility [spark]
cloud-fan opened a new pull request, #54108:
URL: https://github.com/apache/spark/pull/54108
### What changes were proposed in this pull request?
This is a followup to #52402 that addresses backward compatibility concerns:
1. Keep the original `implicit SQLContext` factory methods for full backward compatibility
2. Add new overloads with explicit `SparkSession` parameter for new code
3. Fix `TestGraphRegistrationContext` to provide implicit `spark` and `sqlContext` to avoid name shadowing issues in nested classes
4. Remove redundant `implicit val sparkSession` declarations from pipeline tests that are no longer needed with the fix
### Why are the changes needed?
PR #52402 changed the MemoryStream API to use `implicit SparkSession` which broke backward compatibility for code that only has `implicit SQLContext` available. This followup ensures:
- Old code continues to work without modification
- New code can use SparkSession with explicit parameters
- Internal implementation uses SparkSession (modernization from #52402)
### Does this PR introduce _any_ user-facing change?
No. This maintains full backward compatibility while adding new API options.
### How was this patch tested?
Existing tests pass. The API changes are additive.
### Was this patch authored or co-authored using generative AI tooling?
Yes
Made with [Cursor](https://cursor.com)
--
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-55337][SS] Fix MemoryStream backward compatibility [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #54108:
URL: https://github.com/apache/spark/pull/54108#issuecomment-3867471680
all tests passed, merging to master/4.1, thanks for the review!
--
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-55337][SS] Fix MemoryStream backward compatibility [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on PR #54108:
URL: https://github.com/apache/spark/pull/54108#issuecomment-3849553160
Shall we rerun the CI? It's good to try again before looking into CI failure and say it's not relevant to this change.
--
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-53656][SS][FOLLOWUP] Improve MemoryStream backward compatibility [spark]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #54108:
URL: https://github.com/apache/spark/pull/54108#issuecomment-3839139034
## JIRA Issue Information
=== Task SPARK-53656 ===
Summary: Refactor MemoryStream to use SparkSession instead of SQLContext
Assignee: Ganesha S
Status: Resolved
Affected: ["4.0.1"]
---
*This comment was automatically generated by GitHub Actions*
--
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-55337][SS] Fix MemoryStream backward compatibility [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #54108:
URL: https://github.com/apache/spark/pull/54108#discussion_r2763593037
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/memory.scala:
##########
@@ -43,36 +43,41 @@ import org.apache.spark.sql.internal.connector.SimpleTableProvider
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.util.CaseInsensitiveStringMap
-object MemoryStream extends LowPriorityMemoryStreamImplicits {
+object MemoryStream {
protected val currentBlockId = new AtomicInteger(0)
protected val memoryStreamId = new AtomicInteger(0)
- def apply[A : Encoder](implicit sparkSession: SparkSession): MemoryStream[A] =
- new MemoryStream[A](memoryStreamId.getAndIncrement(), sparkSession)
-
- def apply[A : Encoder](numPartitions: Int)(implicit sparkSession: SparkSession): MemoryStream[A] =
- new MemoryStream[A](memoryStreamId.getAndIncrement(), sparkSession, Some(numPartitions))
-}
-
-/**
- * Provides lower-priority implicits for MemoryStream to prevent ambiguity when both
- * SparkSession and SQLContext are in scope. The implicits in the companion object,
- * which use SparkSession, take higher precedence.
- */
-trait LowPriorityMemoryStreamImplicits {
- this: MemoryStream.type =>
-
- // Deprecated: Used when an implicit SQLContext is in scope
- @deprecated("Use MemoryStream.apply with an implicit SparkSession instead of SQLContext", "4.1.0")
- def apply[A: Encoder]()(implicit sqlContext: SQLContext): MemoryStream[A] =
+ /**
+ * Creates a MemoryStream with an implicit SQLContext (backward compatible).
+ * Usage: `MemoryStream[Int]`
+ */
+ def apply[A: Encoder](implicit sqlContext: SQLContext): MemoryStream[A] =
new MemoryStream[A](memoryStreamId.getAndIncrement(), sqlContext.sparkSession)
- @deprecated("Use MemoryStream.apply with an implicit SparkSession instead of SQLContext", "4.1.0")
- def apply[A: Encoder](numPartitions: Int)(implicit sqlContext: SQLContext): MemoryStream[A] =
+ /**
+ * Creates a MemoryStream with specified partitions using implicit SQLContext.
+ * Usage: `MemoryStream[Int](numPartitions)`
+ */
+ def apply[A: Encoder](numPartitions: Int)(
+ implicit sqlContext: SQLContext): MemoryStream[A] =
new MemoryStream[A](
memoryStreamId.getAndIncrement(),
sqlContext.sparkSession,
Some(numPartitions))
+
+ /**
+ * Creates a MemoryStream with explicit SparkSession.
+ * Usage: `MemoryStream[Int](spark)`
+ */
+ def apply[A: Encoder](sparkSession: SparkSession): MemoryStream[A] =
+ new MemoryStream[A](memoryStreamId.getAndIncrement(), sparkSession)
+
+ /**
+ * Creates a MemoryStream with explicit encoder and SparkSession.
+ * Usage: `MemoryStream(Encoders.scalaInt, spark)`
+ */
+ def apply[A](encoder: Encoder[A], sparkSession: SparkSession): MemoryStream[A] =
Review Comment:
I roughly remember the intention was to discourage the usage of SQLContext - if that's the case, we probably want to have the way to pass `numPartitions` parameter.
That said, looks like this method (explicit encoder instance) is newly added. Is there any usage of this? We don't seem to add the same in `ContinuousMemoryStream` and `LowLatencyMemoryStream`.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/MemorySinkSuite.scala:
##########
@@ -343,84 +343,6 @@ class MemorySinkSuite extends StreamTest with BeforeAndAfter {
intsToDF(expected)(schema))
}
- test("LowPriorityMemoryStreamImplicits works with implicit sqlContext") {
Review Comment:
lol we added tests in wrong place... the file seems to be for memory "sink", not memory "source".
--
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-55337][SS] Fix MemoryStream backward compatibility [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan closed pull request #54108: [SPARK-55337][SS] Fix MemoryStream backward compatibility
URL: https://github.com/apache/spark/pull/54108
--
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-55337][SS] Fix MemoryStream backward compatibility [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #54108:
URL: https://github.com/apache/spark/pull/54108#issuecomment-3848500087
> Do we know about the impact of this change? It seems to be complicated to think through the impact of this change (and a prior change, since it was turned out to be not backward compatible).
It's internal API so this is not strictly a bug fix. It will break new Spark 4.1 app that start to use the new `def apply` with SparkSession implicit, but not fixing it will break a lot more Spark apps that are not upgraded to 4.1 yet. There is no way to support both as they just conflict with each other.
--
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-55337][SS] Fix MemoryStream backward compatibility [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #54108:
URL: https://github.com/apache/spark/pull/54108#discussion_r2764950996
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/memory.scala:
##########
@@ -43,36 +43,41 @@ import org.apache.spark.sql.internal.connector.SimpleTableProvider
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.util.CaseInsensitiveStringMap
-object MemoryStream extends LowPriorityMemoryStreamImplicits {
+object MemoryStream {
protected val currentBlockId = new AtomicInteger(0)
protected val memoryStreamId = new AtomicInteger(0)
- def apply[A : Encoder](implicit sparkSession: SparkSession): MemoryStream[A] =
- new MemoryStream[A](memoryStreamId.getAndIncrement(), sparkSession)
-
- def apply[A : Encoder](numPartitions: Int)(implicit sparkSession: SparkSession): MemoryStream[A] =
- new MemoryStream[A](memoryStreamId.getAndIncrement(), sparkSession, Some(numPartitions))
-}
-
-/**
- * Provides lower-priority implicits for MemoryStream to prevent ambiguity when both
- * SparkSession and SQLContext are in scope. The implicits in the companion object,
- * which use SparkSession, take higher precedence.
- */
-trait LowPriorityMemoryStreamImplicits {
- this: MemoryStream.type =>
-
- // Deprecated: Used when an implicit SQLContext is in scope
- @deprecated("Use MemoryStream.apply with an implicit SparkSession instead of SQLContext", "4.1.0")
- def apply[A: Encoder]()(implicit sqlContext: SQLContext): MemoryStream[A] =
+ /**
+ * Creates a MemoryStream with an implicit SQLContext (backward compatible).
+ * Usage: `MemoryStream[Int]`
+ */
+ def apply[A: Encoder](implicit sqlContext: SQLContext): MemoryStream[A] =
new MemoryStream[A](memoryStreamId.getAndIncrement(), sqlContext.sparkSession)
- @deprecated("Use MemoryStream.apply with an implicit SparkSession instead of SQLContext", "4.1.0")
- def apply[A: Encoder](numPartitions: Int)(implicit sqlContext: SQLContext): MemoryStream[A] =
+ /**
+ * Creates a MemoryStream with specified partitions using implicit SQLContext.
+ * Usage: `MemoryStream[Int](numPartitions)`
+ */
+ def apply[A: Encoder](numPartitions: Int)(
+ implicit sqlContext: SQLContext): MemoryStream[A] =
new MemoryStream[A](
memoryStreamId.getAndIncrement(),
sqlContext.sparkSession,
Some(numPartitions))
+
+ /**
+ * Creates a MemoryStream with explicit SparkSession.
+ * Usage: `MemoryStream[Int](spark)`
+ */
+ def apply[A: Encoder](sparkSession: SparkSession): MemoryStream[A] =
+ new MemoryStream[A](memoryStreamId.getAndIncrement(), sparkSession)
+
+ /**
+ * Creates a MemoryStream with explicit encoder and SparkSession.
+ * Usage: `MemoryStream(Encoders.scalaInt, spark)`
+ */
+ def apply[A](encoder: Encoder[A], sparkSession: SparkSession): MemoryStream[A] =
Review Comment:
I've add a new overload to specific numPartitions with SparkSession.
--
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-53656][SS][FOLLOWUP] Improve MemoryStream backward compatibility [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #54108:
URL: https://github.com/apache/spark/pull/54108#issuecomment-3839235408
cc @ganeshashree @HeartSaVioR
--
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-55337][SS] Fix MemoryStream backward compatibility [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #54108:
URL: https://github.com/apache/spark/pull/54108#discussion_r2764776372
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/memory.scala:
##########
@@ -43,36 +43,41 @@ import org.apache.spark.sql.internal.connector.SimpleTableProvider
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.util.CaseInsensitiveStringMap
-object MemoryStream extends LowPriorityMemoryStreamImplicits {
+object MemoryStream {
protected val currentBlockId = new AtomicInteger(0)
protected val memoryStreamId = new AtomicInteger(0)
- def apply[A : Encoder](implicit sparkSession: SparkSession): MemoryStream[A] =
- new MemoryStream[A](memoryStreamId.getAndIncrement(), sparkSession)
-
- def apply[A : Encoder](numPartitions: Int)(implicit sparkSession: SparkSession): MemoryStream[A] =
- new MemoryStream[A](memoryStreamId.getAndIncrement(), sparkSession, Some(numPartitions))
-}
-
-/**
- * Provides lower-priority implicits for MemoryStream to prevent ambiguity when both
- * SparkSession and SQLContext are in scope. The implicits in the companion object,
- * which use SparkSession, take higher precedence.
- */
-trait LowPriorityMemoryStreamImplicits {
- this: MemoryStream.type =>
-
- // Deprecated: Used when an implicit SQLContext is in scope
- @deprecated("Use MemoryStream.apply with an implicit SparkSession instead of SQLContext", "4.1.0")
- def apply[A: Encoder]()(implicit sqlContext: SQLContext): MemoryStream[A] =
+ /**
+ * Creates a MemoryStream with an implicit SQLContext (backward compatible).
+ * Usage: `MemoryStream[Int]`
+ */
+ def apply[A: Encoder](implicit sqlContext: SQLContext): MemoryStream[A] =
new MemoryStream[A](memoryStreamId.getAndIncrement(), sqlContext.sparkSession)
- @deprecated("Use MemoryStream.apply with an implicit SparkSession instead of SQLContext", "4.1.0")
- def apply[A: Encoder](numPartitions: Int)(implicit sqlContext: SQLContext): MemoryStream[A] =
+ /**
+ * Creates a MemoryStream with specified partitions using implicit SQLContext.
+ * Usage: `MemoryStream[Int](numPartitions)`
+ */
+ def apply[A: Encoder](numPartitions: Int)(
+ implicit sqlContext: SQLContext): MemoryStream[A] =
new MemoryStream[A](
memoryStreamId.getAndIncrement(),
sqlContext.sparkSession,
Some(numPartitions))
+
+ /**
+ * Creates a MemoryStream with explicit SparkSession.
+ * Usage: `MemoryStream[Int](spark)`
+ */
+ def apply[A: Encoder](sparkSession: SparkSession): MemoryStream[A] =
+ new MemoryStream[A](memoryStreamId.getAndIncrement(), sparkSession)
+
+ /**
+ * Creates a MemoryStream with explicit encoder and SparkSession.
+ * Usage: `MemoryStream(Encoders.scalaInt, spark)`
+ */
+ def apply[A](encoder: Encoder[A], sparkSession: SparkSession): MemoryStream[A] =
Review Comment:
It's used in quite some places like `StreamingQueryManagerSuite`
```
testQuietly("can start a streaming query with the same name in a different session") {
val session2 = spark.cloneSession()
val ds1 = MemoryStream(Encoders.INT, spark).toDS()
val ds2 = MemoryStream(Encoders.INT, session2).toDS()
```
--
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-55337][SS] Fix MemoryStream backward compatibility [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on PR #54108:
URL: https://github.com/apache/spark/pull/54108#issuecomment-3840360817
Thank you for getting a new JIRA ID.
--
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-53656][SS][FOLLOWUP] Improve MemoryStream backward compatibility [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #54108:
URL: https://github.com/apache/spark/pull/54108#discussion_r2757288732
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/memory.scala:
##########
@@ -43,36 +43,42 @@ import org.apache.spark.sql.internal.connector.SimpleTableProvider
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.util.CaseInsensitiveStringMap
-object MemoryStream extends LowPriorityMemoryStreamImplicits {
+object MemoryStream {
protected val currentBlockId = new AtomicInteger(0)
protected val memoryStreamId = new AtomicInteger(0)
- def apply[A : Encoder](implicit sparkSession: SparkSession): MemoryStream[A] =
- new MemoryStream[A](memoryStreamId.getAndIncrement(), sparkSession)
-
- def apply[A : Encoder](numPartitions: Int)(implicit sparkSession: SparkSession): MemoryStream[A] =
- new MemoryStream[A](memoryStreamId.getAndIncrement(), sparkSession, Some(numPartitions))
-}
-
-/**
- * Provides lower-priority implicits for MemoryStream to prevent ambiguity when both
- * SparkSession and SQLContext are in scope. The implicits in the companion object,
- * which use SparkSession, take higher precedence.
- */
-trait LowPriorityMemoryStreamImplicits {
- this: MemoryStream.type =>
-
- // Deprecated: Used when an implicit SQLContext is in scope
- @deprecated("Use MemoryStream.apply with an implicit SparkSession instead of SQLContext", "4.1.0")
- def apply[A: Encoder]()(implicit sqlContext: SQLContext): MemoryStream[A] =
Review Comment:
This is the problem. This is not backward compatible, as the previous API is `def apply[A: Encoder](implicit sqlContext: SQLContext): MemoryStream[A]` (no parentheses).
There is no way to keep both implicits. So the proposal here is to only keep implicit `SQLContext`, and require to pass `SparkSession` implicitly.
--
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