You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "pengzhon-db (via GitHub)" <gi...@apache.org> on 2023/05/11 22:40:30 UTC
[GitHub] [spark] pengzhon-db opened a new pull request, #41146: Spark connect function to create dataframe ref
pengzhon-db opened a new pull request, #41146:
URL: https://github.com/apache/spark/pull/41146
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html
2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][SPARK-XXXX] Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a faster review.
7. If you want to add a new configuration, please read the guideline first for naming configurations in
'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
8. If you want to add or modify an error type or message, please read the guideline first in
'core/src/main/resources/error/README.md'.
-->
### What changes were proposed in this pull request?
<!--
Please clarify what changes you are proposing. The purpose of this section is to outline the changes and how this PR fixes the issue.
If possible, please consider writing useful notes for better and faster reviews in your PR. See the examples below.
1. If you refactor some codes with changing classes, showing the class hierarchy will help reviewers.
2. If you fix some SQL features, you can provide some references of other DBMSes.
3. If there is design documentation, please add the link.
4. If there is a discussion in the mailing list, please add the link.
-->
### Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
1. If you propose a new API, clarify the use case for a new API.
2. If you fix a bug, you can clarify why it is a bug.
-->
### Does this PR introduce _any_ user-facing change?
<!--
Note that it means *any* user-facing change including all aspects such as the documentation fix.
If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description and/or an example to show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change compared to the released Spark versions or within the unreleased branches such as master.
If no, write 'No'.
-->
### How was this patch tested?
<!--
If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible.
If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future.
If tests were not added, please describe why they were not added and/or why it was difficult to add.
If benchmark tests were added, please run the benchmarks in GitHub Actions for the consistent environment, and the instructions could accord to: https://spark.apache.org/developer-tools.html#github-workflow-benchmarks.
-->
--
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
[GitHub] [spark] rangadi commented on pull request #41146: [SPARK-43474] [SS] [CONNECT] Add a spark connect function to create DataFrame reference
Posted by "rangadi (via GitHub)" <gi...@apache.org>.
rangadi commented on PR #41146:
URL: https://github.com/apache/spark/pull/41146#issuecomment-1590470175
Please note that updates to this PR are in another PR: https://github.com/apache/spark/pull/41580
--
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
[GitHub] [spark] rangadi commented on a diff in pull request #41146: [SPARK-43474] [SS] [CONNECT] Add a spark connect function to create DataFrame reference
Posted by "rangadi (via GitHub)" <gi...@apache.org>.
rangadi commented on code in PR #41146:
URL: https://github.com/apache/spark/pull/41146#discussion_r1228848876
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectCachedDataFrameManager.scala:
##########
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.connect.service
+
+import scala.collection.mutable
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.DataFrame
+import org.apache.spark.sql.connect.common.InvalidPlanInput
+
+/**
+ * This class caches DataFrame on the server side with given ids. The Spark Connect client can
+ * create a DataFrame reference with the id. When server transforms the DataFrame reference, it
+ * finds the DataFrame from the cache and replace the reference.
+ *
+ * Each (userId, sessionId) has a corresponding DataFrame map. A cached DataFrame can only be
+ * accessed from the same user within the same session. The DataFrame will be removed from the
+ * cache when the session expires.
+ */
+private[connect] class SparkConnectCachedDataFrameManager extends Logging {
+
+ // Each (userId, sessionId) has a DataFrame cache map.
+ private val dataFrameCache = mutable.Map[(String, String), mutable.Map[String, DataFrame]]()
+
+ def put(userId: String, sessionId: String, dataFrameId: String, value: DataFrame): Unit =
+ synchronized {
+ val sessionKey = (userId, sessionId)
+ val sessionDataFrameMap = dataFrameCache
+ .getOrElseUpdate(sessionKey, mutable.Map[String, DataFrame]())
+ sessionDataFrameMap.put(dataFrameId, value)
+ }
Review Comment:
Agree, we could user ConcurrentHashMap. But I often end up preferring `synchronized` as well. Since this is not perf critical (used only for certain DFs), though I am not sure if there is any perf difference.
Added `@GuardedBy` annotation.
See the the continuation of this PR here: https://github.com/apache/spark/pull/41580/files#diff-1a8933e9723f5497c3991441c7ff21fe43db63d483354af9a0113043ea600b3eR42
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectCachedDataFrameManager.scala:
##########
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.connect.service
+
+import scala.collection.mutable
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.DataFrame
+import org.apache.spark.sql.connect.common.InvalidPlanInput
+
+/**
+ * This class caches DataFrame on the server side with given ids. The Spark Connect client can
+ * create a DataFrame reference with the id. When server transforms the DataFrame reference, it
+ * finds the DataFrame from the cache and replace the reference.
+ *
+ * Each (userId, sessionId) has a corresponding DataFrame map. A cached DataFrame can only be
+ * accessed from the same user within the same session. The DataFrame will be removed from the
+ * cache when the session expires.
+ */
+private[connect] class SparkConnectCachedDataFrameManager extends Logging {
Review Comment:
Discussed above. SessionHolder is not accessible yet. Also removed session_id and user_id from this cache, instead making it key on actual Spark session (user_id & session_id is implicit in that)
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkConnectPlanner.scala:
##########
@@ -786,6 +788,12 @@ class SparkConnectPlanner(val session: SparkSession) {
.logicalPlan
}
+ private def transformCachedRemoteRelation(rel: proto.CachedRemoteRelation): LogicalPlan = {
+ SparkConnectService.cachedDataFrameManager
+ .get(rel.getUserId, rel.getSessionId, rel.getRelationId)
+ .logicalPlan
+ }
Review Comment:
Agree. For now proposing to keep it in separate class. Continue discussion [here](https://github.com/apache/spark/pull/41580#discussion_r1228842072).
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectCachedDataFrameManager.scala:
##########
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.connect.service
+
+import scala.collection.mutable
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.DataFrame
+import org.apache.spark.sql.connect.common.InvalidPlanInput
+
+/**
+ * This class caches DataFrame on the server side with given ids. The Spark Connect client can
+ * create a DataFrame reference with the id. When server transforms the DataFrame reference, it
+ * finds the DataFrame from the cache and replace the reference.
+ *
+ * Each (userId, sessionId) has a corresponding DataFrame map. A cached DataFrame can only be
+ * accessed from the same user within the same session. The DataFrame will be removed from the
+ * cache when the session expires.
+ */
+private[connect] class SparkConnectCachedDataFrameManager extends Logging {
Review Comment:
Removed. See continuation of this PR : #41580
##########
python/pyspark/sql/connect/session.py:
##########
@@ -476,6 +476,11 @@ def createDataFrame(
createDataFrame.__doc__ = PySparkSession.createDataFrame.__doc__
+ def _createCachedDataFrame(self, relationId: str) -> "DataFrame":
Review Comment:
Removed. It will used in foreachBatch implementation (in follow up PRs)
##########
connector/connect/common/src/main/protobuf/spark/connect/relations.proto:
##########
@@ -395,6 +396,18 @@ message CachedLocalRelation {
string hash = 3;
}
+// Represents a remote relation that has been cached on server.
+message CachedRemoteRelation {
+ // (Required) An identifier of the user which cached the relation
+ string userId = 1;
+
+ // (Required) An identifier of the Spark session in which the relation is cached
+ string sessionId = 2;
Review Comment:
Agree. This is important. Changed the implementation to use SparkSession as the key (it has as `sessionUUID`)
[continue the discussion [here](https://github.com/apache/spark/pull/41580#discussion_r1228837983)]
--
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
[GitHub] [spark] amaliujia commented on a diff in pull request #41146: [SPARK-43474] [SS] [CONNECT] Add a spark connect function to create DataFrame reference
Posted by "amaliujia (via GitHub)" <gi...@apache.org>.
amaliujia commented on code in PR #41146:
URL: https://github.com/apache/spark/pull/41146#discussion_r1227334505
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectCachedDataFrameManager.scala:
##########
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.connect.service
+
+import scala.collection.mutable
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.DataFrame
+import org.apache.spark.sql.connect.common.InvalidPlanInput
+
+/**
+ * This class caches DataFrame on the server side with given ids. The Spark Connect client can
+ * create a DataFrame reference with the id. When server transforms the DataFrame reference, it
+ * finds the DataFrame from the cache and replace the reference.
+ *
+ * Each (userId, sessionId) has a corresponding DataFrame map. A cached DataFrame can only be
+ * accessed from the same user within the same session. The DataFrame will be removed from the
+ * cache when the session expires.
+ */
+private[connect] class SparkConnectCachedDataFrameManager extends Logging {
+
+ // Each (userId, sessionId) has a DataFrame cache map.
+ private val dataFrameCache = mutable.Map[(String, String), mutable.Map[String, DataFrame]]()
+
+ def put(userId: String, sessionId: String, dataFrameId: String, value: DataFrame): Unit =
+ synchronized {
+ val sessionKey = (userId, sessionId)
+ val sessionDataFrameMap = dataFrameCache
+ .getOrElseUpdate(sessionKey, mutable.Map[String, DataFrame]())
+ sessionDataFrameMap.put(dataFrameId, value)
+ }
Review Comment:
only my personal taste:
I feel like `synchronized` is easy to reason compared to `ConcurrentHashMap` for code readers. Unless there is significantly performance gain somehow if we switch to a concurrent data structure.
--
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
[GitHub] [spark] pengzhon-db commented on pull request #41146: [SPARK-43474] [SS] [CONNECT] Add a spark connect function to create DataFrame reference
Posted by "pengzhon-db (via GitHub)" <gi...@apache.org>.
pengzhon-db commented on PR #41146:
URL: https://github.com/apache/spark/pull/41146#issuecomment-1581113769
@rangadi can u review this PR?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] amaliujia commented on a diff in pull request #41146: [SPARK-43474] [SS] [CONNECT] Add a spark connect function to create DataFrame reference
Posted by "amaliujia (via GitHub)" <gi...@apache.org>.
amaliujia commented on code in PR #41146:
URL: https://github.com/apache/spark/pull/41146#discussion_r1227326437
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectCachedDataFrameManager.scala:
##########
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.connect.service
+
+import scala.collection.mutable
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.DataFrame
+import org.apache.spark.sql.connect.common.InvalidPlanInput
+
+/**
+ * This class caches DataFrame on the server side with given ids. The Spark Connect client can
+ * create a DataFrame reference with the id. When server transforms the DataFrame reference, it
+ * finds the DataFrame from the cache and replace the reference.
+ *
+ * Each (userId, sessionId) has a corresponding DataFrame map. A cached DataFrame can only be
+ * accessed from the same user within the same session. The DataFrame will be removed from the
+ * cache when the session expires.
+ */
+private[connect] class SparkConnectCachedDataFrameManager extends Logging {
Review Comment:
nit: do we need Logging? It is not used?
--
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
[GitHub] [spark] rangadi commented on a diff in pull request #41146: [SPARK-43474] [SS] [CONNECT] Add a spark connect function to create DataFrame reference
Posted by "rangadi (via GitHub)" <gi...@apache.org>.
rangadi commented on code in PR #41146:
URL: https://github.com/apache/spark/pull/41146#discussion_r1224627406
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectService.scala:
##########
@@ -298,12 +298,15 @@ object SparkConnectService {
userSessionMapping.getIfPresent((userId, sessionId))
})
+ private[connect] val cachedDataFrameManager = new SparkConnectCachedDataFrameManager()
+
private class RemoveSessionListener extends RemovalListener[SessionCacheKey, SessionHolder] {
override def onRemoval(
notification: RemovalNotification[SessionCacheKey, SessionHolder]): Unit = {
val SessionHolder(userId, sessionId, session) = notification.getValue
val blockManager = session.sparkContext.env.blockManager
blockManager.removeCache(userId, sessionId)
+ cachedDataFrameManager.remove(userId, sessionId)
Review Comment:
Note to self: This reference should be removed from streaming engine once the foreach batch completes..
--
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
[GitHub] [spark] grundprinzip commented on a diff in pull request #41146: [SPARK-43474] [SS] [CONNECT] Add a spark connect function to create DataFrame reference
Posted by "grundprinzip (via GitHub)" <gi...@apache.org>.
grundprinzip commented on code in PR #41146:
URL: https://github.com/apache/spark/pull/41146#discussion_r1228238923
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkConnectPlanner.scala:
##########
@@ -786,6 +788,12 @@ class SparkConnectPlanner(val session: SparkSession) {
.logicalPlan
}
+ private def transformCachedRemoteRelation(rel: proto.CachedRemoteRelation): LogicalPlan = {
+ SparkConnectService.cachedDataFrameManager
+ .get(rel.getUserId, rel.getSessionId, rel.getRelationId)
+ .logicalPlan
+ }
Review Comment:
Conceptually, the cached data should come from the session holder that could be passed to the planner instead.
##########
connector/connect/common/src/main/protobuf/spark/connect/relations.proto:
##########
@@ -395,6 +396,18 @@ message CachedLocalRelation {
string hash = 3;
}
+// Represents a remote relation that has been cached on server.
+message CachedRemoteRelation {
+ // (Required) An identifier of the user which cached the relation
+ string userId = 1;
+
+ // (Required) An identifier of the Spark session in which the relation is cached
+ string sessionId = 2;
Review Comment:
The user, session ID can't be trusted coming from the proto. THe cached relation must only have the actual unique ID of the relation ID and the rest is resolved from the context of the query.
##########
python/pyspark/sql/connect/session.py:
##########
@@ -476,6 +476,11 @@ def createDataFrame(
createDataFrame.__doc__ = PySparkSession.createDataFrame.__doc__
+ def _createCachedDataFrame(self, relationId: str) -> "DataFrame":
Review Comment:
this seems to be unused here?
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectCachedDataFrameManager.scala:
##########
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.connect.service
+
+import scala.collection.mutable
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.DataFrame
+import org.apache.spark.sql.connect.common.InvalidPlanInput
+
+/**
+ * This class caches DataFrame on the server side with given ids. The Spark Connect client can
+ * create a DataFrame reference with the id. When server transforms the DataFrame reference, it
+ * finds the DataFrame from the cache and replace the reference.
+ *
+ * Each (userId, sessionId) has a corresponding DataFrame map. A cached DataFrame can only be
+ * accessed from the same user within the same session. The DataFrame will be removed from the
+ * cache when the session expires.
+ */
+private[connect] class SparkConnectCachedDataFrameManager extends Logging {
+
+ // Each (userId, sessionId) has a DataFrame cache map.
+ private val dataFrameCache = mutable.Map[(String, String), mutable.Map[String, DataFrame]]()
+
+ def put(userId: String, sessionId: String, dataFrameId: String, value: DataFrame): Unit =
+ synchronized {
+ val sessionKey = (userId, sessionId)
+ val sessionDataFrameMap = dataFrameCache
+ .getOrElseUpdate(sessionKey, mutable.Map[String, DataFrame]())
+ sessionDataFrameMap.put(dataFrameId, value)
+ }
Review Comment:
This will make this easier as well because you only have one concurrent map
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectCachedDataFrameManager.scala:
##########
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.connect.service
+
+import scala.collection.mutable
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.DataFrame
+import org.apache.spark.sql.connect.common.InvalidPlanInput
+
+/**
+ * This class caches DataFrame on the server side with given ids. The Spark Connect client can
+ * create a DataFrame reference with the id. When server transforms the DataFrame reference, it
+ * finds the DataFrame from the cache and replace the reference.
+ *
+ * Each (userId, sessionId) has a corresponding DataFrame map. A cached DataFrame can only be
+ * accessed from the same user within the same session. The DataFrame will be removed from the
+ * cache when the session expires.
+ */
+private[connect] class SparkConnectCachedDataFrameManager extends Logging {
Review Comment:
Can we add this class to the session holder to make sure that this is properly associated to the right user ID and session.
--
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
[GitHub] [spark] rangadi commented on a diff in pull request #41146: [SPARK-43474] [SS] [CONNECT] Add a spark connect function to create DataFrame reference
Posted by "rangadi (via GitHub)" <gi...@apache.org>.
rangadi commented on code in PR #41146:
URL: https://github.com/apache/spark/pull/41146#discussion_r1223516152
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectCachedDataFrameManager.scala:
##########
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.connect.service
+
+import scala.collection.mutable
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.DataFrame
+import org.apache.spark.sql.connect.common.InvalidPlanInput
+
+/**
+ * This class caches DataFrame on the server side with a given key as id. The Spark Connect client
+ * can create a DataFrame reference with the key. When server transforms the DataFrame reference,
+ * it finds the DataFrame from the cache and replace the reference.
+ *
+ * Each (userId, sessionId) has a corresponding DataFrame map. A cached DataFrame can only be
+ * accessed from the same user within the same session. The DataFrame will be removed from the cache
+ * when the session expires.
+ */
+private[connect] class SparkConnectCachedDataFrameManager extends Logging {
+
+ // Each (userId, sessionId) has a DataFrame cache map.
+ private val dataFrameCache = mutable.Map[(String, String), mutable.Map[String, DataFrame]]()
+
+ def put(userId: String, sessionId: String, key: String, value: DataFrame): Unit = synchronized {
Review Comment:
Better to rename `key` as `dataFrameId`.
##########
connector/connect/common/src/main/protobuf/spark/connect/relations.proto:
##########
@@ -394,6 +395,18 @@ message CachedLocalRelation {
string hash = 3;
}
+// Represents a DataFrame that has been cached on server.
+message CachedDataFrame {
Review Comment:
How about renaming this `CachedRemoteRelation`? DataFrame is an API level concept.
##########
connector/connect/common/src/main/protobuf/spark/connect/relations.proto:
##########
@@ -394,6 +395,18 @@ message CachedLocalRelation {
string hash = 3;
}
+// Represents a DataFrame that has been cached on server.
+message CachedDataFrame {
+ // (Required) An identifier of the user which cached the dataframe
+ string userId = 1;
+
+ // (Required) An identifier of the Spark session in which the dataframe is cached
+ string sessionId = 2;
+
+ // (Required) A key represents the id of the cached dataframe
+ string key = 3;
Review Comment:
Better to rename this as `id`, `relationId`, or `remoteId`.
--
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
[GitHub] [spark] pengzhon-db commented on a diff in pull request #41146: [SPARK-43474] [SS] [CONNECT] Add a spark connect function to create DataFrame reference
Posted by "pengzhon-db (via GitHub)" <gi...@apache.org>.
pengzhon-db commented on code in PR #41146:
URL: https://github.com/apache/spark/pull/41146#discussion_r1224639307
##########
connector/connect/common/src/main/protobuf/spark/connect/relations.proto:
##########
@@ -394,6 +395,18 @@ message CachedLocalRelation {
string hash = 3;
}
+// Represents a DataFrame that has been cached on server.
+message CachedDataFrame {
+ // (Required) An identifier of the user which cached the dataframe
+ string userId = 1;
+
+ // (Required) An identifier of the Spark session in which the dataframe is cached
+ string sessionId = 2;
+
+ // (Required) A key represents the id of the cached dataframe
+ string key = 3;
Review Comment:
update to relationId
--
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
[GitHub] [spark] hvanhovell closed pull request #41146: [SPARK-43474] [SS] [CONNECT] Add a spark connect function to create DataFrame reference
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell closed pull request #41146: [SPARK-43474] [SS] [CONNECT] Add a spark connect function to create DataFrame reference
URL: https://github.com/apache/spark/pull/41146
--
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
[GitHub] [spark] pengzhon-db commented on a diff in pull request #41146: [SPARK-43474] [SS] [CONNECT] Add a spark connect function to create DataFrame reference
Posted by "pengzhon-db (via GitHub)" <gi...@apache.org>.
pengzhon-db commented on code in PR #41146:
URL: https://github.com/apache/spark/pull/41146#discussion_r1191808469
##########
connector/connect/common/src/main/protobuf/spark/connect/relations.proto:
##########
@@ -394,6 +395,18 @@ message CachedLocalRelation {
string hash = 3;
}
+// Represents a DataFrame that has been cached on server.
+message CachedDataFrame {
+ // (Required) An identifier of the user which cached the dataframe
+ string userId = 1;
Review Comment:
We can also just get userId and sessionId from server via request, instead of passing from here.
But that would require we update [transformRelation()](https://github.com/apache/spark/blob/master/connector/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkConnectPlanner.scala#L87) to take into two more parameters, which means all all those `transform...()` need to be updated to have two more parameter.
--
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
[GitHub] [spark] zhenlineo commented on a diff in pull request #41146: [SPARK-43474] [SS] [CONNECT] Add a spark connect function to create DataFrame reference
Posted by "zhenlineo (via GitHub)" <gi...@apache.org>.
zhenlineo commented on code in PR #41146:
URL: https://github.com/apache/spark/pull/41146#discussion_r1227327242
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectCachedDataFrameManager.scala:
##########
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.connect.service
+
+import scala.collection.mutable
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.DataFrame
+import org.apache.spark.sql.connect.common.InvalidPlanInput
+
+/**
+ * This class caches DataFrame on the server side with given ids. The Spark Connect client can
+ * create a DataFrame reference with the id. When server transforms the DataFrame reference, it
+ * finds the DataFrame from the cache and replace the reference.
+ *
+ * Each (userId, sessionId) has a corresponding DataFrame map. A cached DataFrame can only be
+ * accessed from the same user within the same session. The DataFrame will be removed from the
+ * cache when the session expires.
+ */
+private[connect] class SparkConnectCachedDataFrameManager extends Logging {
+
+ // Each (userId, sessionId) has a DataFrame cache map.
+ private val dataFrameCache = mutable.Map[(String, String), mutable.Map[String, DataFrame]]()
+
+ def put(userId: String, sessionId: String, dataFrameId: String, value: DataFrame): Unit =
+ synchronized {
+ val sessionKey = (userId, sessionId)
+ val sessionDataFrameMap = dataFrameCache
+ .getOrElseUpdate(sessionKey, mutable.Map[String, DataFrame]())
+ sessionDataFrameMap.put(dataFrameId, value)
+ }
Review Comment:
How about using two concurrent hash maps + `compute` to avoid `synchronized`? For example:
```
dataFrameCache.compute(sessionKey, (key, sessionDataFrameMap) => {
val newMap = if (sessionDataFrameMap == null) new ConcurrentHashMap[String, String]() else sessionDataFrameMap
newMap.put(dataFrameId, value)
})
```
Similar logics apply for `remove`.
For get, you just need to get without the need to lock.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] zhenlineo commented on a diff in pull request #41146: [SPARK-43474] [SS] [CONNECT] Add a spark connect function to create DataFrame reference
Posted by "zhenlineo (via GitHub)" <gi...@apache.org>.
zhenlineo commented on code in PR #41146:
URL: https://github.com/apache/spark/pull/41146#discussion_r1227327242
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectCachedDataFrameManager.scala:
##########
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.connect.service
+
+import scala.collection.mutable
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.DataFrame
+import org.apache.spark.sql.connect.common.InvalidPlanInput
+
+/**
+ * This class caches DataFrame on the server side with given ids. The Spark Connect client can
+ * create a DataFrame reference with the id. When server transforms the DataFrame reference, it
+ * finds the DataFrame from the cache and replace the reference.
+ *
+ * Each (userId, sessionId) has a corresponding DataFrame map. A cached DataFrame can only be
+ * accessed from the same user within the same session. The DataFrame will be removed from the
+ * cache when the session expires.
+ */
+private[connect] class SparkConnectCachedDataFrameManager extends Logging {
+
+ // Each (userId, sessionId) has a DataFrame cache map.
+ private val dataFrameCache = mutable.Map[(String, String), mutable.Map[String, DataFrame]]()
+
+ def put(userId: String, sessionId: String, dataFrameId: String, value: DataFrame): Unit =
+ synchronized {
+ val sessionKey = (userId, sessionId)
+ val sessionDataFrameMap = dataFrameCache
+ .getOrElseUpdate(sessionKey, mutable.Map[String, DataFrame]())
+ sessionDataFrameMap.put(dataFrameId, value)
+ }
Review Comment:
How about using two concurrent hash maps + `compute` to avoid `synchronized`? For example:
```
dataFrameCache.compute(sessionKey, (key, sessionDataFrameMap) => {
val newMap = if (sessionDataFrameMap == null) new ConcurrentHashMap[String, String]() else sessionDataFrameMap
newMap.put(dataFrameId, value)
})
```
Similar logics apply for `remove`.
For get, you just need to get without the need to explicitly lock.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org