[KYUUBI #3227] SparkConfParser supports parse bytes and time

### _Why are the changes needed?_

SparkConfParser supports parse bytes and time for better user experience.

### _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 #3227 from pan3793/tpc.

Closes #3227

a2d8847 [Cheng Pan] style
87e54f5 [Cheng Pan] review
381316e [Cheng Pan] nit
5a23f0e [Cheng Pan] fix
48dfe33 [Cheng Pan] nit
c7c85f7 [Cheng Pan] SparkConfParser support parse bytes and time

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

Author: pan3793

docs/connector/spark/tpch.rst

@@ -75,7 +75,7 @@ To add TPC-H tables as a catalog, we can set the following configurations in ``$
    spark.sql.catalog.tpch.useAnsiStringType=false
 
    # (optional) Maximum bytes per task, consider reducing it if you want higher parallelism.
-   spark.sql.catalog.tpch.read.maxPartitionBytes=134217728
+   spark.sql.catalog.tpch.read.maxPartitionBytes=128m
 
 TPC-H Operations
 ----------------

extensions/spark/kyuubi-spark-connector-common/pom.xml

@@ -66,16 +66,27 @@
         </dependency>
 
         <dependency>
-            <groupId>org.scalatestplus</groupId>
-            <artifactId>scalacheck-1-15_${scala.binary.version}</artifactId>
+            <groupId>org.apache.spark</groupId>
+            <artifactId>spark-sql_${scala.binary.version}</artifactId>
+            <type>test-jar</type>
             <scope>test</scope>
         </dependency>
 
         <dependency>
-            <groupId>org.apache.spark</groupId>
-            <artifactId>spark-sql_${scala.binary.version}</artifactId>
-            <version>${spark.version}</version>
-            <type>test-jar</type>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-client-api</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-client-runtime</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.scalatestplus</groupId>
+            <artifactId>scalacheck-1-15_${scala.binary.version}</artifactId>
             <scope>test</scope>
         </dependency>
     </dependencies>

extensions/spark/kyuubi-spark-connector-common/src/main/java/org/apache/kyuubi/spark/connector/common/JavaUtils.java

@@ -0,0 +1,184 @@
+/*
+ * 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.common;
+
+import java.util.HashMap;
+import java.util.Locale;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import org.apache.spark.network.util.ByteUnit;
+
+/** Copied from Apache Spark org.apache.spark.network.util.JavaUtils */
+public class JavaUtils {
+
+  private static final Map<String, TimeUnit> timeSuffixes;
+
+  static {
+    timeSuffixes = new HashMap<>();
+    timeSuffixes.put("us", TimeUnit.MICROSECONDS);
+    timeSuffixes.put("ms", TimeUnit.MILLISECONDS);
+    timeSuffixes.put("s", TimeUnit.SECONDS);
+    timeSuffixes.put("m", TimeUnit.MINUTES);
+    timeSuffixes.put("min", TimeUnit.MINUTES);
+    timeSuffixes.put("h", TimeUnit.HOURS);
+    timeSuffixes.put("d", TimeUnit.DAYS);
+  }
+
+  private static final Map<String, ByteUnit> byteSuffixes;
+
+  static {
+    byteSuffixes = new HashMap<>();
+    byteSuffixes.put("b", ByteUnit.BYTE);
+    byteSuffixes.put("k", ByteUnit.KiB);
+    byteSuffixes.put("kb", ByteUnit.KiB);
+    byteSuffixes.put("m", ByteUnit.MiB);
+    byteSuffixes.put("mb", ByteUnit.MiB);
+    byteSuffixes.put("g", ByteUnit.GiB);
+    byteSuffixes.put("gb", ByteUnit.GiB);
+    byteSuffixes.put("t", ByteUnit.TiB);
+    byteSuffixes.put("tb", ByteUnit.TiB);
+    byteSuffixes.put("p", ByteUnit.PiB);
+    byteSuffixes.put("pb", ByteUnit.PiB);
+  }
+
+  /**
+   * Convert a passed time string (e.g. 50s, 100ms, or 250us) to a time count in the given unit. The
+   * unit is also considered the default if the given string does not specify a unit.
+   */
+  public static long timeStringAs(String str, TimeUnit unit) {
+    String lower = str.toLowerCase(Locale.ROOT).trim();
+
+    try {
+      Matcher m = Pattern.compile("(-?[0-9]+)([a-z]+)?").matcher(lower);
+      if (!m.matches()) {
+        throw new NumberFormatException("Failed to parse time string: " + str);
+      }
+
+      long val = Long.parseLong(m.group(1));
+      String suffix = m.group(2);
+
+      // Check for invalid suffixes
+      if (suffix != null && !timeSuffixes.containsKey(suffix)) {
+        throw new NumberFormatException("Invalid suffix: \"" + suffix + "\"");
+      }
+
+      // If suffix is valid use that, otherwise none was provided and use the default passed
+      return unit.convert(val, suffix != null ? timeSuffixes.get(suffix) : unit);
+    } catch (NumberFormatException e) {
+      String timeError =
+          "Time must be specified as seconds (s), "
+              + "milliseconds (ms), microseconds (us), minutes (m or min), hour (h), or day (d). "
+              + "E.g. 50s, 100ms, or 250us.";
+
+      throw new NumberFormatException(timeError + "\n" + e.getMessage());
+    }
+  }
+
+  /**
+   * Convert a time parameter such as (50s, 100ms, or 250us) to milliseconds for internal use. If no
+   * suffix is provided, the passed number is assumed to be in ms.
+   */
+  public static long timeStringAsMs(String str) {
+    return timeStringAs(str, TimeUnit.MILLISECONDS);
+  }
+
+  /**
+   * Convert a time parameter such as (50s, 100ms, or 250us) to seconds for internal use. If no
+   * suffix is provided, the passed number is assumed to be in seconds.
+   */
+  public static long timeStringAsSec(String str) {
+    return timeStringAs(str, TimeUnit.SECONDS);
+  }
+
+  /**
+   * Convert a passed byte string (e.g. 50b, 100kb, or 250mb) to the given. If no suffix is
+   * provided, a direct conversion to the provided unit is attempted.
+   */
+  public static long byteStringAs(String str, ByteUnit unit) {
+    String lower = str.toLowerCase(Locale.ROOT).trim();
+
+    try {
+      Matcher m = Pattern.compile("([0-9]+)([a-z]+)?").matcher(lower);
+      Matcher fractionMatcher = Pattern.compile("([0-9]+\\.[0-9]+)([a-z]+)?").matcher(lower);
+
+      if (m.matches()) {
+        long val = Long.parseLong(m.group(1));
+        String suffix = m.group(2);
+
+        // Check for invalid suffixes
+        if (suffix != null && !byteSuffixes.containsKey(suffix)) {
+          throw new NumberFormatException("Invalid suffix: \"" + suffix + "\"");
+        }
+
+        // If suffix is valid use that, otherwise none was provided and use the default passed
+        return unit.convertFrom(val, suffix != null ? byteSuffixes.get(suffix) : unit);
+      } else if (fractionMatcher.matches()) {
+        throw new NumberFormatException(
+            "Fractional values are not supported. Input was: " + fractionMatcher.group(1));
+      } else {
+        throw new NumberFormatException("Failed to parse byte string: " + str);
+      }
+
+    } catch (NumberFormatException e) {
+      String byteError =
+          "Size must be specified as bytes (b), "
+              + "kibibytes (k), mebibytes (m), gibibytes (g), tebibytes (t), or pebibytes(p). "
+              + "E.g. 50b, 100k, or 250m.";
+
+      throw new NumberFormatException(byteError + "\n" + e.getMessage());
+    }
+  }
+
+  /**
+   * Convert a passed byte string (e.g. 50b, 100k, or 250m) to bytes for internal use.
+   *
+   * <p>If no suffix is provided, the passed number is assumed to be in bytes.
+   */
+  public static long byteStringAsBytes(String str) {
+    return byteStringAs(str, ByteUnit.BYTE);
+  }
+
+  /**
+   * Convert a passed byte string (e.g. 50b, 100k, or 250m) to kibibytes for internal use.
+   *
+   * <p>If no suffix is provided, the passed number is assumed to be in kibibytes.
+   */
+  public static long byteStringAsKb(String str) {
+    return byteStringAs(str, ByteUnit.KiB);
+  }
+
+  /**
+   * Convert a passed byte string (e.g. 50b, 100k, or 250m) to mebibytes for internal use.
+   *
+   * <p>If no suffix is provided, the passed number is assumed to be in mebibytes.
+   */
+  public static long byteStringAsMb(String str) {
+    return byteStringAs(str, ByteUnit.MiB);
+  }
+
+  /**
+   * Convert a passed byte string (e.g. 50b, 100k, or 250m) to gibibytes for internal use.
+   *
+   * <p>If no suffix is provided, the passed number is assumed to be in gibibytes.
+   */
+  public static long byteStringAsGb(String str) {
+    return byteStringAs(str, ByteUnit.GiB);
+  }
+}

extensions/spark/kyuubi-spark-connector-common/src/main/scala/org/apache/kyuubi/spark/connector/common/SparkConfParser.scala

@@ -38,6 +38,8 @@ case class SparkConfParser(
   def longConf(): LongConfParser = new LongConfParser()
   def doubleConf(): DoubleConfParser = new DoubleConfParser()
   def stringConf(): StringConfParser = new StringConfParser()
+  def bytesConf(): BytesConfParser = new BytesConfParser()
+  def timeConf(): TimeConfParser = new TimeConfParser()
 
   class BooleanConfParser extends ConfParser[Boolean] {
     override protected def conversion(value: String): Boolean = value.toBoolean
@@ -59,11 +61,19 @@ case class SparkConfParser(
     override protected def conversion(value: String): String = value
   }
 
+  class BytesConfParser extends ConfParser[Long] {
+    override protected def conversion(value: String): Long = JavaUtils.byteStringAsBytes(value)
+  }
+
+  class TimeConfParser extends ConfParser[Long] {
+    override protected def conversion(value: String): Long = JavaUtils.timeStringAsMs(value)
+  }
+
   abstract class ConfParser[T]() {
     private var optionName: Option[String] = None
     private var sessionConfName: Option[String] = None
     private var tablePropertyName: Option[String] = None
-    private var defaultValue: Option[T] = None
+    private var defaultStringValue: Option[String] = None
 
     def option(name: String): ConfParser[T] = {
       this.optionName = Some(name)
@@ -80,8 +90,8 @@ case class SparkConfParser(
       this
     }
 
-    def defaultValue(value: T): ConfParser[T] = {
-      this.defaultValue = Some(value)
+    def defaultStringValue(value: String): ConfParser[T] = {
+      this.defaultStringValue = Some(value)
       this
     }
 
@@ -96,18 +106,13 @@ case class SparkConfParser(
       if (valueOpt.isEmpty && properties != null) {
         valueOpt = tablePropertyName.flatMap(name => Option(properties.get(name)))
       }
-      valueOpt = valueOpt.filter(_ != null)
-      if (valueOpt.isDefined) {
-        valueOpt.map(conversion(_))
-      } else {
-        defaultValue
-      }
+      valueOpt.orElse(defaultStringValue).map(conversion)
     }
 
     protected def conversion(value: String): T
 
     def parse(): T = {
-      assert(defaultValue.isDefined, "Default value cannot be empty.")
+      assert(defaultStringValue.isDefined, "Default value cannot be empty.")
       parseOptional().get
     }
 

extensions/spark/kyuubi-spark-connector-common/src/test/scala/org/apache/kyuubi/spark/connector/common/SparkConfParserSuite.scala

@@ -17,49 +17,50 @@
 
 package org.apache.kyuubi.spark.connector.common
 
-// scalastyle:off anyfunsuite
 import scala.collection.JavaConverters._
 
+import org.apache.spark.SparkFunSuite
 import org.apache.spark.sql.util.CaseInsensitiveStringMap
-import org.scalatest.BeforeAndAfterAll
-import org.scalatest.funsuite.AnyFunSuite
 
-class SparkConfParserSuite extends AnyFunSuite with BeforeAndAfterAll {
-// scalastyle:on anyfunsuite
+class SparkConfParserSuite extends SparkFunSuite {
 
-  test("parse options config") {
-    assert(confParser.stringConf().option("optKey1").defaultValue("test").parse() === "optValue1")
-    assert(confParser.booleanConf().option("booleanKey").defaultValue(true).parse() === false)
-    assert(confParser.intConf().option("intKey").defaultValue(0).parse() === 10)
-    assert(confParser.longConf().option("longKey").defaultValue(0).parse() === Long.MaxValue)
-    assert(confParser.doubleConf().option("doubleKey").defaultValue(0.0).parse() === 1.1)
-  }
-
-  test("parse properties config") {
-    assert(confParser.intConf().option("key1")
-      .tableProperty("key1")
-      .defaultValue(0).parse() === 111)
-    assert(confParser.stringConf()
-      .option("propertyKey1")
-      .tableProperty("propertyKey1")
-      .defaultValue("test").parse() === "propertyValue1")
-  }
-
-  private var confParser: SparkConfParser = null
+  private var confParser: SparkConfParser = _
 
   override protected def beforeAll(): Unit = {
     super.beforeAll()
     val options = new CaseInsensitiveStringMap(Map(
       "key1" -> "111",
-      "optKey1" -> "optValue1",
       "booleanKey" -> "false",
       "intKey" -> "10",
       "longKey" -> String.valueOf(Long.MaxValue),
-      "doubleKey" -> "1.1").asJava)
+      "doubleKey" -> "1.1",
+      "bytesKey" -> "1k",
+      "timeKey" -> "1s").asJava)
     val properties = Map(
       "key1" -> "333",
       "propertyKey1" -> "propertyValue1")
     confParser = SparkConfParser(options, null, properties.asJava)
   }
 
+  test("parse options config") {
+    assert(confParser.stringConf().option("optKey1").defaultStringValue("test").parse() === "test")
+    assert(
+      confParser.booleanConf().option("booleanKey").defaultStringValue("true").parse() === false)
+    assert(confParser.intConf().option("intKey").defaultStringValue("0").parse() === 10)
+    assert(
+      confParser.longConf().option("longKey").defaultStringValue("0").parse() === Long.MaxValue)
+    assert(confParser.doubleConf().option("doubleKey").defaultStringValue("0.0").parse() === 1.1)
+    assert(confParser.bytesConf().option("bytesKey").defaultStringValue("0k").parse() === 1024L)
+    assert(confParser.timeConf().option("timeKey").defaultStringValue("0s").parse() === 1000L)
+  }
+
+  test("parse properties config") {
+    assert(confParser.intConf().option("key1")
+      .tableProperty("key1")
+      .defaultStringValue("0").parse() === 111)
+    assert(confParser.stringConf()
+      .option("propertyKey1")
+      .tableProperty("propertyKey1")
+      .defaultStringValue("test").parse() === "propertyValue1")
+  }
 }