Repository: carbondata Updated Branches: refs/heads/branch-1.3 4544765a4 -> 28f33746a
[CARBONDATA-2329] Non Serializable extra info in session is overwritten from stale thread Problem: 1. Non Serializable extra info is copied from thread which causes stale data from old session when the thread is reused by spark. 2. CarboSessionInfo clone is not copying Non serializable info to new object which can damage session level values if local query thread updates values. Solution: Remove logic to copy Non Serializable extra info and fix clone logic This closes #2154 Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/28f33746 Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/28f33746 Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/28f33746 Branch: refs/heads/branch-1.3 Commit: 28f33746a2feb8774cf40e8e392e9980c64d0292 Parents: 4544765 Author: KanakaKumar <kanaka.avv...@huawei.com> Authored: Tue Apr 10 19:42:30 2018 +0530 Committer: manishgupta88 <tomanishgupt...@gmail.com> Committed: Wed Apr 11 14:33:44 2018 +0530 ---------------------------------------------------------------------- .../org/apache/carbondata/core/util/CarbonSessionInfo.java | 6 ++---- .../spark2/src/main/scala/org/apache/spark/sql/CarbonEnv.scala | 2 -- .../src/main/scala/org/apache/spark/sql/CarbonSession.scala | 4 ---- 3 files changed, 2 insertions(+), 10 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/carbondata/blob/28f33746/core/src/main/java/org/apache/carbondata/core/util/CarbonSessionInfo.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/carbondata/core/util/CarbonSessionInfo.java b/core/src/main/java/org/apache/carbondata/core/util/CarbonSessionInfo.java index e397e55..3da44be 100644 --- a/core/src/main/java/org/apache/carbondata/core/util/CarbonSessionInfo.java +++ b/core/src/main/java/org/apache/carbondata/core/util/CarbonSessionInfo.java @@ -74,11 +74,9 @@ public class CarbonSessionInfo implements Serializable, Cloneable { ex.printStackTrace(); } } - Map<String, Object> nonSerializableExtraInfo = getNonSerializableExtraInfo(); - for (Map.Entry<String, Object> entry : nonSerializableExtraInfo.entrySet()) { - nonSerializableExtraInfo.put(entry.getKey(), entry.getValue()); + for (Map.Entry<String, Object> entry : getNonSerializableExtraInfo().entrySet()) { + newObj.getNonSerializableExtraInfo().put(entry.getKey(), entry.getValue()); } - newObj.setNonSerializableExtraInfo(nonSerializableExtraInfo); return newObj; } http://git-wip-us.apache.org/repos/asf/carbondata/blob/28f33746/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonEnv.scala ---------------------------------------------------------------------- diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonEnv.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonEnv.scala index 4647bee..ed7e83b 100644 --- a/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonEnv.scala +++ b/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonEnv.scala @@ -75,11 +75,9 @@ class CarbonEnv { // update carbon session parameters , preserve thread parameters val currentThreadSesssionInfo = ThreadLocalSessionInfo.getCarbonSessionInfo carbonSessionInfo = new CarbonSessionInfo() - sessionParams = carbonSessionInfo.getSessionParams if (currentThreadSesssionInfo != null) { carbonSessionInfo.setThreadParams(currentThreadSesssionInfo.getThreadParams) } - carbonSessionInfo.setSessionParams(sessionParams) ThreadLocalSessionInfo.setCarbonSessionInfo(carbonSessionInfo) val config = new CarbonSQLConf(sparkSession) if (sparkSession.conf.getOption(CarbonCommonConstants.ENABLE_UNSAFE_SORT).isEmpty) { http://git-wip-us.apache.org/repos/asf/carbondata/blob/28f33746/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonSession.scala ---------------------------------------------------------------------- diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonSession.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonSession.scala index 0116d9e..e95b8db 100644 --- a/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonSession.scala +++ b/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonSession.scala @@ -246,11 +246,7 @@ object CarbonSession { // copy all the thread parameters to apply to session parameters currentThreadSessionInfo.getThreadParams.getAll.asScala .foreach(entry => carbonSessionInfo.getSessionParams.addProperty(entry._1, entry._2)) - currentThreadSessionInfo.getNonSerializableExtraInfo.asScala - .foreach(entry => carbonSessionInfo.getNonSerializableExtraInfo.put(entry._1, entry._2)) carbonSessionInfo.setThreadParams(currentThreadSessionInfo.getThreadParams) - carbonSessionInfo - .setNonSerializableExtraInfo(currentThreadSessionInfo.getNonSerializableExtraInfo) } // preserve thread parameters across call ThreadLocalSessionInfo.setCarbonSessionInfo(carbonSessionInfo)
