[GLUTEN-9244][CORE] Change the way of passing default timezone to native config

cpp/core/config/GlutenConfig.h

@@ -37,8 +37,6 @@ const std::string kAllowPrecisionLoss = "spark.sql.decimalOperations.allowPrecis
 
 const std::string kIgnoreMissingFiles = "spark.sql.files.ignoreMissingFiles";
 
-const std::string kDefaultSessionTimezone = "spark.gluten.sql.session.timeZone.default";
-
 const std::string kSparkOverheadMemory = "spark.gluten.memoryOverhead.size.in.bytes";
 
 const std::string kSparkOffHeapMemory = "spark.gluten.memory.offHeap.size.in.bytes";

cpp/velox/compute/WholeStageResultIterator.cc

@@ -477,12 +477,7 @@ std::unordered_map<std::string, std::string> WholeStageResultIterator::getQueryC
   configs[velox::core::QueryConfig::kMaxOutputBatchRows] =
       std::to_string(veloxCfg_->get<uint32_t>(kSparkBatchSize, 4096));
   try {
-    if (veloxCfg_->valueExists(kDefaultSessionTimezone)) {
-      configs[velox::core::QueryConfig::kSessionTimezone] = veloxCfg_->get<std::string>(kDefaultSessionTimezone, "");
-    }
-    if (veloxCfg_->valueExists(kSessionTimezone)) {
-      configs[velox::core::QueryConfig::kSessionTimezone] = veloxCfg_->get<std::string>(kSessionTimezone, "");
-    }
+    configs[velox::core::QueryConfig::kSessionTimezone] = veloxCfg_->get<std::string>(kSessionTimezone, "");
     // Adjust timestamp according to the above configured session timezone.
     configs[velox::core::QueryConfig::kAdjustTimestampToTimezone] = "true";
 

gluten-core/src/main/scala/org/apache/gluten/GlutenPlugin.scala

@@ -171,10 +171,6 @@ private[gluten] class GlutenDriverPlugin extends DriverPlugin with Logging {
           s"and set '$SPARK_OFFHEAP_SIZE_KEY' to be greater than $minOffHeapSize")
     }
 
-    // Session's local time zone must be set. If not explicitly set by user, its default
-    // value (detected for the platform) is used, consistent with spark.
-    conf.set(GLUTEN_DEFAULT_SESSION_TIMEZONE.key, SQLConf.SESSION_LOCAL_TIMEZONE.defaultValueString)
-
     // Task slots.
     val taskSlots = SparkResourceUtil.getTaskSlots(conf)
     conf.set(NUM_TASK_SLOTS_PER_EXECUTOR.key, taskSlots.toString)

gluten-ut/spark32/src/test/scala/org/apache/spark/sql/catalyst/expressions/GlutenDateExpressionsSuite.scala

@@ -432,8 +432,7 @@ class GlutenDateExpressionsSuite extends DateExpressionsSuite with GlutenTestsTr
       Seq("legacy", "corrected").foreach {
         legacyParserPolicy =>
           withSQLConf(
-            SQLConf.LEGACY_TIME_PARSER_POLICY.key -> legacyParserPolicy,
-            SQLConf.SESSION_LOCAL_TIMEZONE.key -> UTC_OPT.get
+            SQLConf.LEGACY_TIME_PARSER_POLICY.key -> legacyParserPolicy
           ) {
             assert(Hour(Literal.create(null, DateType), UTC_OPT).resolved === false)
             assert(Hour(Literal(ts), UTC_OPT).resolved)

gluten-ut/spark33/src/test/scala/org/apache/spark/sql/catalyst/expressions/GlutenDateExpressionsSuite.scala

@@ -430,8 +430,7 @@ class GlutenDateExpressionsSuite extends DateExpressionsSuite with GlutenTestsTr
       Seq("legacy", "corrected").foreach {
         legacyParserPolicy =>
           withSQLConf(
-            SQLConf.LEGACY_TIME_PARSER_POLICY.key -> legacyParserPolicy,
-            SQLConf.SESSION_LOCAL_TIMEZONE.key -> UTC_OPT.get
+            SQLConf.LEGACY_TIME_PARSER_POLICY.key -> legacyParserPolicy
           ) {
             assert(Hour(Literal.create(null, DateType), UTC_OPT).resolved === false)
             assert(Hour(Literal(ts), UTC_OPT).resolved)

gluten-ut/spark34/src/test/scala/org/apache/spark/sql/catalyst/expressions/GlutenDateExpressionsSuite.scala

@@ -430,8 +430,7 @@ class GlutenDateExpressionsSuite extends DateExpressionsSuite with GlutenTestsTr
       Seq("legacy", "corrected").foreach {
         legacyParserPolicy =>
           withSQLConf(
-            SQLConf.LEGACY_TIME_PARSER_POLICY.key -> legacyParserPolicy,
-            SQLConf.SESSION_LOCAL_TIMEZONE.key -> UTC_OPT.get
+            SQLConf.LEGACY_TIME_PARSER_POLICY.key -> legacyParserPolicy
           ) {
             assert(Hour(Literal.create(null, DateType), UTC_OPT).resolved === false)
             assert(Hour(Literal(ts), UTC_OPT).resolved)

gluten-ut/spark35/src/test/scala/org/apache/spark/sql/catalyst/expressions/GlutenDateExpressionsSuite.scala

@@ -430,8 +430,7 @@ class GlutenDateExpressionsSuite extends DateExpressionsSuite with GlutenTestsTr
       Seq("legacy", "corrected").foreach {
         legacyParserPolicy =>
           withSQLConf(
-            SQLConf.LEGACY_TIME_PARSER_POLICY.key -> legacyParserPolicy,
-            SQLConf.SESSION_LOCAL_TIMEZONE.key -> UTC_OPT.get
+            SQLConf.LEGACY_TIME_PARSER_POLICY.key -> legacyParserPolicy
           ) {
             assert(Hour(Literal.create(null, DateType), UTC_OPT).resolved === false)
             assert(Hour(Literal(ts), UTC_OPT).resolved)

shims/common/src/main/scala/org/apache/gluten/config/GlutenConfig.scala

@@ -461,8 +461,6 @@ object GlutenConfig {
       COLUMNAR_TASK_OFFHEAP_SIZE_IN_BYTES.key,
       COLUMNAR_MAX_BATCH_SIZE.key,
       SHUFFLE_WRITER_BUFFER_SIZE.key,
-      SQLConf.SESSION_LOCAL_TIMEZONE.key,
-      GLUTEN_DEFAULT_SESSION_TIMEZONE.key,
       SQLConf.LEGACY_SIZE_OF_NULL.key,
       "spark.io.compression.codec",
       "spark.sql.decimalOperations.allowPrecisionLoss",
@@ -516,7 +514,8 @@ object GlutenConfig {
         GLUTEN_COLUMNAR_TO_ROW_MEM_THRESHOLD.key,
         GLUTEN_COLUMNAR_TO_ROW_MEM_THRESHOLD.defaultValue.get.toString),
       (SPARK_SHUFFLE_SPILL_COMPRESS, SPARK_SHUFFLE_SPILL_COMPRESS_DEFAULT.toString),
-      (SQLConf.MAP_KEY_DEDUP_POLICY.key, SQLConf.MAP_KEY_DEDUP_POLICY.defaultValueString)
+      (SQLConf.MAP_KEY_DEDUP_POLICY.key, SQLConf.MAP_KEY_DEDUP_POLICY.defaultValueString),
+      (SESSION_LOCAL_TIMEZONE.key, SESSION_LOCAL_TIMEZONE.defaultValueString)
     )
     keyWithDefault.forEach(e => nativeConfMap.put(e._1, conf.getOrElse(e._1, e._2)))
     GlutenConfigUtil.mapByteConfValue(
@@ -595,7 +594,8 @@ object GlutenConfig {
       ("spark.gluten.sql.columnar.backend.velox.fileHandleCacheEnabled", "false"),
       ("spark.gluten.velox.awsSdkLogLevel", "FATAL"),
       ("spark.gluten.velox.s3UseProxyFromEnv", "false"),
-      ("spark.gluten.velox.s3PayloadSigningPolicy", "Never")
+      ("spark.gluten.velox.s3PayloadSigningPolicy", "Never"),
+      (SESSION_LOCAL_TIMEZONE.key, SESSION_LOCAL_TIMEZONE.defaultValueString)
     )
     keyWithDefault.forEach(e => nativeConfMap.put(e._1, conf.getOrElse(e._1, e._2)))
 
@@ -609,7 +609,6 @@ object GlutenConfig {
       COLUMNAR_OFFHEAP_SIZE_IN_BYTES.key,
       COLUMNAR_TASK_OFFHEAP_SIZE_IN_BYTES.key,
       SPARK_OFFHEAP_ENABLED,
-      SESSION_LOCAL_TIMEZONE.key,
       DECIMAL_OPERATIONS_ALLOW_PREC_LOSS.key,
       SPARK_REDACTION_REGEX,
       LEGACY_TIME_PARSER_POLICY.key
@@ -655,13 +654,6 @@ object GlutenConfig {
     .booleanConf
     .createWithDefault(true)
 
-  val GLUTEN_DEFAULT_SESSION_TIMEZONE = buildConf("spark.gluten.sql.session.timeZone.default")
-    .doc(
-      "used to hold default session timezone and will be really used by Gluten only if " +
-        "`spark.sql.session.timeZone` is not set.")
-    .stringConf
-    .createWithDefaultString(SQLConf.SESSION_LOCAL_TIMEZONE.defaultValueString)
-
   val GLUTEN_LOAD_LIB_OS =
     buildConf("spark.gluten.loadLibOS")
       .doc("The shared library loader's OS name.")