[KYUUBI #2984] Refactor TPCDS configurations using SparkConfParser

### _Why are the changes needed?_

close #2984

### _How was this patch tested?_
- [ ] Add some test cases that check the changes thoroughly including negative and positive cases if possible

- [ ] Add screenshots for manual tests if appropriate

- [x] [Run test](https://kyuubi.apache.org/docs/latest/develop_tools/testing.html#running-tests) locally before make a pull request

Closes #2988 from Kwafoor/master.

Closes #2984

d4f761d [wangjunbo] [KYUUBI-2984]Refactor TPCDS configurations using SparkConfParser
5525084 [wangjunbo] [KYUUBI-2984]Refactor TPCDS configurations using SparkConfParser
293afc3 [wangjunbo] [KYUUBI-2984]Refactor TPCDS configurations using SparkConfParser
4ce8ba3 [wangjunbo] [KYUUBI-2984]Refactor TPCDS configurations using SparkConfParser

Authored-by: wangjunbo <wangjunbo@qiyi.com>
Signed-off-by: Cheng Pan <chengpan@apache.org>

Author: wangjunbo

extensions/spark/kyuubi-spark-connector-tpcds/src/main/scala/org/apache/kyuubi/spark/connector/tpcds/TPCDSConf.scala

@@ -0,0 +1,53 @@
+/*
+ * 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.kyuubi.spark.connector.tpcds
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+import org.apache.kyuubi.spark.connector.common.SparkConfParser
+import org.apache.kyuubi.spark.connector.tpcds.TPCDSConf._
+
+case class TPCDSConf(spark: SparkSession, options: CaseInsensitiveStringMap) {
+
+  private val confParser: SparkConfParser = SparkConfParser(options, spark.conf, null)
+  // When true, use CHAR VARCHAR; otherwise use STRING
+  lazy val useAnsiStringType: Boolean = confParser.booleanConf()
+    .option(USE_ANSI_STRING_TYPE)
+    .sessionConf(USE_ANSI_STRING_TYPE)
+    .defaultValue(USE_ANSI_STRING_TYPE_DEFAULT)
+    .parse()
+  // 09-26-2017 v2.6.0
+  // Replaced two occurrences of "c_last_review_date" with "c_last_review_date_sk" to be consistent
+  // with Table 2-14 (Customer Table Column Definitions) in section 2.4.7 of the specification
+  // (fogbugz 2046).
+  //
+  // https://www.tpc.org/tpc_documents_current_versions/pdf/tpc-ds_v3.2.0.pdf
+  lazy val useTableSchema_2_6: Boolean = confParser.booleanConf()
+    .option(USE_TABLE_SCHEMA_2_6)
+    .sessionConf(USE_TABLE_SCHEMA_2_6)
+    .defaultValue(USE_TABLE_SCHEMA_2_6_DEFAULT)
+    .parse()
+}
+
+object TPCDSConf {
+  val USE_ANSI_STRING_TYPE = "useAnsiStringType"
+  val USE_ANSI_STRING_TYPE_DEFAULT = false;
+  val USE_TABLE_SCHEMA_2_6 = "useTableSchema_2_6"
+  val USE_TABLE_SCHEMA_2_6_DEFAULT = true
+}

extensions/spark/kyuubi-spark-connector-tpcds/src/main/scala/org/apache/kyuubi/spark/connector/tpcds/TPCDSTable.scala

@@ -35,21 +35,12 @@ import org.apache.spark.sql.util.CaseInsensitiveStringMap
 class TPCDSTable(tbl: String, scale: Double, options: CaseInsensitiveStringMap)
   extends SparkTable with SupportsRead {
 
-  // When true, use CHAR VARCHAR; otherwise use STRING
-  val useAnsiStringType: Boolean = options.getBoolean("useAnsiStringType", false)
-
-  // 09-26-2017 v2.6.0
-  // Replaced two occurrences of "c_last_review_date" with "c_last_review_date_sk" to be consistent
-  // with Table 2-14 (Customer Table Column Definitions) in section 2.4.7 of the specification
-  // (fogbugz 2046).
-  //
-  // https://www.tpc.org/tpc_documents_current_versions/pdf/tpc-ds_v3.2.0.pdf
-  val useTableSchema_2_6: Boolean = options.getBoolean("useTableSchema_2_6", true)
-
   val tpcdsTable: Table = Table.getTable(tbl)
 
   lazy val spark: SparkSession = SparkSession.active
 
+  lazy val tpcdsConf: TPCDSConf = TPCDSConf(spark, options);
+
   override def name: String = s"${TPCDSSchemaUtils.dbName(scale)}.$tbl"
 
   override def toString: String = s"TPCDSTable($name)"
@@ -67,14 +58,14 @@ class TPCDSTable(tbl: String, scale: Double, options: CaseInsensitiveStringMap)
         // Like: io.trino.tpcds.row.CallCenterRow.getValues
         val index = TPCDSSchemaUtils.reviseNullColumnIndex(tpcdsTable, i)
         StructField(
-          TPCDSSchemaUtils.reviseColumnName(c, useTableSchema_2_6),
+          TPCDSSchemaUtils.reviseColumnName(c, tpcdsConf.useTableSchema_2_6),
           toSparkDataType(c.getType),
           nullable(index))
       })
   }
 
   override def partitioning: Array[Transform] = TPCDSSchemaUtils
-    .tablePartitionColumnNames(tpcdsTable, useTableSchema_2_6)
+    .tablePartitionColumnNames(tpcdsTable, tpcdsConf.useTableSchema_2_6)
     .map { Expressions.identity }
 
   override def capabilities(): util.Set[TableCapability] =
@@ -92,9 +83,9 @@ class TPCDSTable(tbl: String, scale: Double, options: CaseInsensitiveStringMap)
       case (DATE, None, None) => DateType
       case (DECIMAL, Some(precision), Some(scale)) => DecimalType(precision, scale)
       case (VARCHAR, Some(precision), None) =>
-        if (useAnsiStringType) VarcharType(precision) else StringType
+        if (tpcdsConf.useAnsiStringType) VarcharType(precision) else StringType
       case (CHAR, Some(precision), None) =>
-        if (useAnsiStringType) CharType(precision) else StringType
+        if (tpcdsConf.useAnsiStringType) CharType(precision) else StringType
       case (t, po, so) =>
         throw new IllegalArgumentException(s"Unsupported TPC-DS type: ($t, $po, $so)")
     }