[KYUUBI #2541] Set nullable in table schema

### _Why are the changes needed?_

close #2541

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

- [ ] Add screenshots for manual tests if appropriate

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

Closes #2635 from wForget/KYUUBI-2541.

Closes #2541

3ec56f7 [wforget] fix
d1a6766 [wforget] refactor and comments
84ccafa [wforget] add comment and rename method
4b8af0e [wforget] revise column index
105e9ad [wforget] [KYUUBI-2541] Set nullable in table schema

Authored-by: wforget <643348094@qq.com>
Signed-off-by: Cheng Pan <chengpan@apache.org>

Author: wForget

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

@@ -22,7 +22,6 @@ import java.util
 import scala.collection.JavaConverters._
 import scala.collection.immutable
 
-import io.trino.tpcds.Table
 import org.apache.spark.sql.catalyst.analysis.{NoSuchNamespaceException, NoSuchTableException}
 import org.apache.spark.sql.connector.catalog.{Identifier, NamespaceChange, SupportsNamespaces, Table => SparkTable, TableCatalog, TableChange}
 import org.apache.spark.sql.connector.expressions.Transform
@@ -31,8 +30,7 @@ import org.apache.spark.sql.util.CaseInsensitiveStringMap
 
 class TPCDSCatalog extends TableCatalog with SupportsNamespaces {
 
-  val tables: Array[String] = Table.getBaseTables.asScala
-    .map(_.getName).filterNot(_ == "dbgen_version").toArray
+  val tables: Array[String] = TPCDSTableUtils.BASE_TABLES.map(_.getName)
 
   val scales: Array[Int] = Array(0, 1, 10, 100, 300, 1000, 3000, 10000, 30000, 100000)
 

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

@@ -61,11 +61,18 @@ class TPCDSTable(tbl: String, scale: Int, options: CaseInsensitiveStringMap)
   override def toString: String = s"TPCDSTable($name)"
 
   override def schema: StructType = {
-    // TODO tpcdsTable.notNullBitMap does not correct, set nullable follows
-    //      https://tpc.org/TPC_Documents_Current_Versions/pdf/TPC-DS_v3.2.0.pdf
+    def nullable(index: Int): Boolean = {
+      val bitMask = 1L << index
+      (bitMask & ~tpcdsTable.getNotNullBitMap) != 0
+    }
     StructType(
       tpcdsTable.getColumns.zipWithIndex.map { case (c, i) =>
-        StructField(reviseColumnName(c), toSparkDataType(c.getType))
+        // Because the order of `GeneratorColumn` and `Column` of some tables is inconsistent,
+        // we need to revise the index of null column, in order to be consistent
+        // with the calculation of null column in the getValues method of Row.
+        // Like: io.trino.tpcds.row.CallCenterRow.getValues
+        val index = TPCDSTableUtils.reviseNullColumnIndex(tpcdsTable, i)
+        StructField(reviseColumnName(c), toSparkDataType(c.getType), nullable(index))
       })
   }