Repository: carbondata Updated Branches: refs/heads/branch-1.3 bc3f82510 -> 15cc7fa97
[CARBONDATA-2182] Added one more params called extraParams in SessionParams and add carbonSessionInfo to CarbonEnvInitPreEvent Add one more param called ExtraParmas in SessionParams for session Level operations and pass the carbonSessionInfo to event, so that user can save information in that at session level in carbonSessionInfo This closes #1978 Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/15cc7fa9 Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/15cc7fa9 Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/15cc7fa9 Branch: refs/heads/branch-1.3 Commit: 15cc7fa97722d055ad5627b3a915ee6d2b6817d6 Parents: bc3f825 Author: akashrn5 <[email protected]> Authored: Wed Feb 14 19:07:15 2018 +0530 Committer: manishgupta88 <[email protected]> Committed: Thu Feb 22 11:43:22 2018 +0530 ---------------------------------------------------------------------- .../carbondata/core/util/CarbonSessionInfo.java | 23 +++++++++++++++++++- .../carbondata/core/util/SessionParams.java | 2 ++ .../carbondata/events/CarbonInitEvents.scala | 4 +++- .../scala/org/apache/spark/sql/CarbonEnv.scala | 2 +- .../org/apache/spark/sql/CarbonSession.scala | 4 ++++ 5 files changed, 32 insertions(+), 3 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/carbondata/blob/15cc7fa9/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 a44bde7..e397e55 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 @@ -18,6 +18,7 @@ package org.apache.carbondata.core.util; import java.io.Serializable; +import java.util.HashMap; import java.util.Map; import org.apache.carbondata.core.exception.InvalidConfigurationException; @@ -27,10 +28,13 @@ import org.apache.carbondata.core.exception.InvalidConfigurationException; */ public class CarbonSessionInfo implements Serializable, Cloneable { - private static final long serialVersionUID = 7738818814501121256L; + private static final long serialVersionUID = 4335254187209416779L; + // contains carbon session param details private SessionParams sessionParams; private SessionParams threadParams; + // use the below field to store the objects which need not be serialized + private transient Map<String, Object> nonSerializableExtraInfo; public SessionParams getSessionParams() { return sessionParams; @@ -70,6 +74,23 @@ 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()); + } + newObj.setNonSerializableExtraInfo(nonSerializableExtraInfo); return newObj; } + + public Map<String, Object> getNonSerializableExtraInfo() { + // as the field is transient it can be null if serialized and de serialized again + if (null == nonSerializableExtraInfo) { + nonSerializableExtraInfo = new HashMap<>(); + } + return nonSerializableExtraInfo; + } + + public void setNonSerializableExtraInfo(Map<String, Object> nonSerializableExtraInfo) { + this.nonSerializableExtraInfo = nonSerializableExtraInfo; + } } http://git-wip-us.apache.org/repos/asf/carbondata/blob/15cc7fa9/core/src/main/java/org/apache/carbondata/core/util/SessionParams.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/carbondata/core/util/SessionParams.java b/core/src/main/java/org/apache/carbondata/core/util/SessionParams.java index a6ff61e..c232b1e 100644 --- a/core/src/main/java/org/apache/carbondata/core/util/SessionParams.java +++ b/core/src/main/java/org/apache/carbondata/core/util/SessionParams.java @@ -50,9 +50,11 @@ public class SessionParams implements Serializable { private static final LogService LOGGER = LogServiceFactory.getLogService(CacheProvider.class.getName()); + private static final long serialVersionUID = -7801994600594915264L; private Map<String, String> sProps; private Map<String, String> addedProps; + // below field to be used when we want the objects to be serialized private Map<String, Object> extraInfo; public SessionParams() { sProps = new HashMap<>(); http://git-wip-us.apache.org/repos/asf/carbondata/blob/15cc7fa9/integration/spark-common/src/main/scala/org/apache/carbondata/events/CarbonInitEvents.scala ---------------------------------------------------------------------- diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/events/CarbonInitEvents.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/events/CarbonInitEvents.scala index b76c534..3b1c6a6 100644 --- a/integration/spark-common/src/main/scala/org/apache/carbondata/events/CarbonInitEvents.scala +++ b/integration/spark-common/src/main/scala/org/apache/carbondata/events/CarbonInitEvents.scala @@ -18,13 +18,15 @@ package org.apache.carbondata.events import org.apache.spark.sql.SparkSession +import org.apache.carbondata.core.util.CarbonSessionInfo + /** * * @param sparkSession * @param storePath */ case class CarbonEnvInitPreEvent( - sparkSession: SparkSession, storePath: String) + sparkSession: SparkSession, carbonSessionInfo: CarbonSessionInfo, storePath: String) extends Event with SessionEventInfo http://git-wip-us.apache.org/repos/asf/carbondata/blob/15cc7fa9/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 8444d25..39e73ee 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 @@ -97,7 +97,7 @@ class CarbonEnv { // trigger event for CarbonEnv init val operationContext = new OperationContext val carbonEnvInitPreEvent: CarbonEnvInitPreEvent = - CarbonEnvInitPreEvent(sparkSession, storePath) + CarbonEnvInitPreEvent(sparkSession, carbonSessionInfo, storePath) OperationListenerBus.getInstance.fireEvent(carbonEnvInitPreEvent, operationContext) CarbonMetaStoreFactory.createCarbonMetaStore(sparkSession.conf) http://git-wip-us.apache.org/repos/asf/carbondata/blob/15cc7fa9/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 e95b8db..0116d9e 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,7 +246,11 @@ 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)
