You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@carbondata.apache.org by aj...@apache.org on 2020/07/27 15:19:23 UTC
[carbondata] branch master updated: [CARBONDATA-3924] Add default
dynamic parameters only one time in one JVM process
This is an automated email from the ASF dual-hosted git repository.
ajantha pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/carbondata.git
The following commit(s) were added to refs/heads/master by this push:
new 226323b [CARBONDATA-3924] Add default dynamic parameters only one time in one JVM process
226323b is described below
commit 226323be5cf4525cc2719bcafe1b86628d6c5db1
Author: QiangCai <qi...@qq.com>
AuthorDate: Mon Jul 27 15:01:25 2020 +0800
[CARBONDATA-3924] Add default dynamic parameters only one time in one JVM process
Why is this PR needed?
PR#3805 introduces a problem that the system will add default dynamic parameters many times in a JVM process at concurrent query case.
If ConfigEntry.registerEntry method registers an exists entry again, it will throw exception.
What changes were proposed in this PR?
Invoking CarbonSQLConf.addDefaultParams only one time in a JVM process
Does this PR introduce any user interface change?
No
Is any new testcase added?
No
This closes #3863
---
.../scala/org/apache/spark/sql/CarbonEnv.scala | 9 +-
.../{CarbonSqlConf.scala => CarbonSQLConf.scala} | 128 +++++++++++----------
.../sql/CarbonGetTableDetailComandTestCase.scala | 0
3 files changed, 69 insertions(+), 68 deletions(-)
diff --git a/integration/spark/src/main/scala/org/apache/spark/sql/CarbonEnv.scala b/integration/spark/src/main/scala/org/apache/spark/sql/CarbonEnv.scala
index 67de334..fbe9e32 100644
--- a/integration/spark/src/main/scala/org/apache/spark/sql/CarbonEnv.scala
+++ b/integration/spark/src/main/scala/org/apache/spark/sql/CarbonEnv.scala
@@ -19,6 +19,7 @@ package org.apache.spark.sql
import java.util.concurrent.ConcurrentHashMap
+import org.apache.spark.internal.config.ConfigEntry
import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.catalyst.analysis.{NoSuchDatabaseException, NoSuchTableException}
import org.apache.spark.sql.catalyst.catalog.SessionCatalog
@@ -113,12 +114,8 @@ class CarbonEnv {
ThreadLocalSessionInfo.setCarbonSessionInfo(threadLevelCarbonSessionInfo)
ThreadLocalSessionInfo.setConfigurationToCurrentThread(
sparkSession.sessionState.newHadoopConf())
- val config = new CarbonSQLConf(sparkSession)
- if (sparkSession.conf.getOption(CarbonCommonConstants.ENABLE_UNSAFE_SORT).isEmpty) {
- config.addDefaultCarbonParams()
- }
- // add session params after adding DefaultCarbonParams
- config.addDefaultCarbonSessionParams()
+ // add default dynamic session params
+ CarbonSQLConf.addDefaultSessionParams(sparkSession);
carbonMetaStore = {
// trigger event for CarbonEnv create
val operationContext = new OperationContext
diff --git a/integration/spark/src/main/scala/org/apache/spark/sql/hive/CarbonSqlConf.scala b/integration/spark/src/main/scala/org/apache/spark/sql/hive/CarbonSQLConf.scala
similarity index 51%
rename from integration/spark/src/main/scala/org/apache/spark/sql/hive/CarbonSqlConf.scala
rename to integration/spark/src/main/scala/org/apache/spark/sql/hive/CarbonSQLConf.scala
index 51f7dff..326ea23 100644
--- a/integration/spark/src/main/scala/org/apache/spark/sql/hive/CarbonSqlConf.scala
+++ b/integration/spark/src/main/scala/org/apache/spark/sql/hive/CarbonSQLConf.scala
@@ -17,6 +17,7 @@
package org.apache.spark.sql.hive
+import org.apache.spark.internal.config.ConfigEntry
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.internal.SQLConf.buildConf
@@ -26,80 +27,83 @@ import org.apache.carbondata.core.util.CarbonProperties
/**
* To initialize dynamic values default param
*/
-class CarbonSQLConf(sparkSession: SparkSession) {
+object CarbonSQLConf {
- val carbonProperties = CarbonProperties.getInstance()
+ private lazy val checkDefaultParams: Boolean = {
+ // add default parameters only one time for a JVM process
+ addDefaultParams()
+ true
+ }
/**
* To initialize dynamic param defaults along with usage docs
*/
- def addDefaultCarbonParams(): Unit = {
- val ENABLE_UNSAFE_SORT =
- buildConf(CarbonCommonConstants.ENABLE_UNSAFE_SORT)
- .doc("To enable/ disable unsafe sort.")
- .booleanConf
- .createWithDefault(carbonProperties.getProperty(CarbonCommonConstants.ENABLE_UNSAFE_SORT,
- CarbonCommonConstants.ENABLE_UNSAFE_SORT_DEFAULT).toBoolean)
- val CARBON_CUSTOM_BLOCK_DISTRIBUTION =
- buildConf(CarbonCommonConstants.CARBON_CUSTOM_BLOCK_DISTRIBUTION)
- .doc("To set carbon task distribution.")
- .stringConf
- .createWithDefault(carbonProperties
- .getProperty(CarbonCommonConstants.CARBON_TASK_DISTRIBUTION,
- CarbonCommonConstants.CARBON_TASK_DISTRIBUTION_DEFAULT))
- val BAD_RECORDS_LOGGER_ENABLE =
- buildConf(CarbonLoadOptionConstants.CARBON_OPTIONS_BAD_RECORDS_LOGGER_ENABLE)
- .doc("To enable/ disable carbon bad record logger.")
- .booleanConf
- .createWithDefault(CarbonLoadOptionConstants
- .CARBON_OPTIONS_BAD_RECORDS_LOGGER_ENABLE_DEFAULT.toBoolean)
- val BAD_RECORDS_ACTION =
- buildConf(CarbonLoadOptionConstants.CARBON_OPTIONS_BAD_RECORDS_ACTION)
- .doc("To configure the bad records action.")
- .stringConf
- .createWithDefault(carbonProperties
- .getProperty(CarbonCommonConstants.CARBON_BAD_RECORDS_ACTION,
- CarbonCommonConstants.CARBON_BAD_RECORDS_ACTION_DEFAULT))
- val IS_EMPTY_DATA_BAD_RECORD =
- buildConf(CarbonLoadOptionConstants.CARBON_OPTIONS_IS_EMPTY_DATA_BAD_RECORD)
- .doc("Property to decide weather empty data to be considered bad/ good record.")
- .booleanConf
- .createWithDefault(CarbonLoadOptionConstants.CARBON_OPTIONS_IS_EMPTY_DATA_BAD_RECORD_DEFAULT
- .toBoolean)
- val SORT_SCOPE =
- buildConf(CarbonLoadOptionConstants.CARBON_OPTIONS_SORT_SCOPE)
- .doc("Property to specify sort scope.")
- .stringConf
- .createWithDefault(carbonProperties.getProperty(CarbonCommonConstants.LOAD_SORT_SCOPE,
- CarbonCommonConstants.LOAD_SORT_SCOPE_DEFAULT))
- val BAD_RECORD_PATH =
- buildConf(CarbonLoadOptionConstants.CARBON_OPTIONS_BAD_RECORD_PATH)
- .doc("Property to configure the bad record location.")
- .stringConf
- .createWithDefault(carbonProperties.getProperty(CarbonCommonConstants.CARBON_BADRECORDS_LOC,
- CarbonCommonConstants.CARBON_BADRECORDS_LOC_DEFAULT_VAL))
- val GLOBAL_SORT_PARTITIONS =
- buildConf(CarbonLoadOptionConstants.CARBON_OPTIONS_GLOBAL_SORT_PARTITIONS)
- .doc("Property to configure the global sort partitions.")
- .stringConf
- .createWithDefault(carbonProperties
- .getProperty(CarbonCommonConstants.LOAD_GLOBAL_SORT_PARTITIONS,
- CarbonCommonConstants.LOAD_GLOBAL_SORT_PARTITIONS_DEFAULT))
- val DATEFORMAT =
- buildConf(CarbonLoadOptionConstants.CARBON_OPTIONS_DATEFORMAT)
- .doc("Property to configure data format for date type columns.")
- .stringConf
- .createWithDefault(CarbonLoadOptionConstants.CARBON_OPTIONS_DATEFORMAT_DEFAULT)
- val CARBON_INPUT_SEGMENTS = buildConf(
- "carbon.input.segments.<database_name>.<table_name>")
+ private def addDefaultParams(): Unit = {
+ if (ConfigEntry.findEntry(CarbonCommonConstants.ENABLE_UNSAFE_SORT) != null) {
+ return
+ }
+ val carbonProperties = CarbonProperties.getInstance()
+ buildConf(CarbonCommonConstants.ENABLE_UNSAFE_SORT)
+ .doc("To enable/ disable unsafe sort.")
+ .booleanConf
+ .createWithDefault(carbonProperties.getProperty(CarbonCommonConstants.ENABLE_UNSAFE_SORT,
+ CarbonCommonConstants.ENABLE_UNSAFE_SORT_DEFAULT).toBoolean)
+ buildConf(CarbonCommonConstants.CARBON_CUSTOM_BLOCK_DISTRIBUTION)
+ .doc("To set carbon task distribution.")
+ .stringConf
+ .createWithDefault(carbonProperties
+ .getProperty(CarbonCommonConstants.CARBON_TASK_DISTRIBUTION,
+ CarbonCommonConstants.CARBON_TASK_DISTRIBUTION_DEFAULT))
+ buildConf(CarbonLoadOptionConstants.CARBON_OPTIONS_BAD_RECORDS_LOGGER_ENABLE)
+ .doc("To enable/ disable carbon bad record logger.")
+ .booleanConf
+ .createWithDefault(CarbonLoadOptionConstants
+ .CARBON_OPTIONS_BAD_RECORDS_LOGGER_ENABLE_DEFAULT.toBoolean)
+ buildConf(CarbonLoadOptionConstants.CARBON_OPTIONS_BAD_RECORDS_ACTION)
+ .doc("To configure the bad records action.")
+ .stringConf
+ .createWithDefault(carbonProperties
+ .getProperty(CarbonCommonConstants.CARBON_BAD_RECORDS_ACTION,
+ CarbonCommonConstants.CARBON_BAD_RECORDS_ACTION_DEFAULT))
+ buildConf(CarbonLoadOptionConstants.CARBON_OPTIONS_IS_EMPTY_DATA_BAD_RECORD)
+ .doc("Property to decide weather empty data to be considered bad/ good record.")
+ .booleanConf
+ .createWithDefault(CarbonLoadOptionConstants.CARBON_OPTIONS_IS_EMPTY_DATA_BAD_RECORD_DEFAULT
+ .toBoolean)
+ buildConf(CarbonLoadOptionConstants.CARBON_OPTIONS_SORT_SCOPE)
+ .doc("Property to specify sort scope.")
+ .stringConf
+ .createWithDefault(carbonProperties.getProperty(CarbonCommonConstants.LOAD_SORT_SCOPE,
+ CarbonCommonConstants.LOAD_SORT_SCOPE_DEFAULT))
+ buildConf(CarbonLoadOptionConstants.CARBON_OPTIONS_BAD_RECORD_PATH)
+ .doc("Property to configure the bad record location.")
+ .stringConf
+ .createWithDefault(carbonProperties.getProperty(CarbonCommonConstants.CARBON_BADRECORDS_LOC,
+ CarbonCommonConstants.CARBON_BADRECORDS_LOC_DEFAULT_VAL))
+ buildConf(CarbonLoadOptionConstants.CARBON_OPTIONS_GLOBAL_SORT_PARTITIONS)
+ .doc("Property to configure the global sort partitions.")
+ .stringConf
+ .createWithDefault(carbonProperties
+ .getProperty(CarbonCommonConstants.LOAD_GLOBAL_SORT_PARTITIONS,
+ CarbonCommonConstants.LOAD_GLOBAL_SORT_PARTITIONS_DEFAULT))
+ buildConf(CarbonLoadOptionConstants.CARBON_OPTIONS_DATEFORMAT)
+ .doc("Property to configure data format for date type columns.")
+ .stringConf
+ .createWithDefault(CarbonLoadOptionConstants.CARBON_OPTIONS_DATEFORMAT_DEFAULT)
+ buildConf("carbon.input.segments.<database_name>.<table_name>")
.doc("Property to configure the list of segments to query.").stringConf
.createWithDefault(carbonProperties
.getProperty("carbon.input.segments.<database_name>.<table_name>", "*"))
}
+
/**
* to set the dynamic properties default values
*/
- def addDefaultCarbonSessionParams(): Unit = {
+ def addDefaultSessionParams(sparkSession: SparkSession): Unit = {
+ // add default global parameters at first
+ CarbonSQLConf.checkDefaultParams
+ // add default session parameters
+ val carbonProperties = CarbonProperties.getInstance()
sparkSession.conf.set(CarbonCommonConstants.ENABLE_UNSAFE_SORT,
carbonProperties.getProperty(CarbonCommonConstants.ENABLE_UNSAFE_SORT,
CarbonCommonConstants.ENABLE_UNSAFE_SORT_DEFAULT).toBoolean)
diff --git a/integration/spark/src/test/scala/org/apache/spark/sql/CarbonGetTableDetailComandTestCase.scala b/integration/spark/src/test/scala/org/apache/spark/sql/CarbonGetTableDetailComandTestCase.scala
deleted file mode 100644
index e69de29..0000000